如何在Spark Streaming中对HDFS流数据按指定键聚合求和？

你的需求是对同一单词且同一文档编号的行，将中间四列分数求和，现有代码的核心问题是没正确设置分组的key，且没有处理字符串转数值的问题，下面是修正后的完整实现：

from pyspark.streaming import StreamingContext
import time
import pprint
from pyspark.sql import functions as F

# 初始化StreamingContext，批次间隔60秒
ssc = StreamingContext(sc, 60)
# 监听HDFS目录下的流数据
lines = ssc.textFileStream("hdfs://thesis:9000/user/kush/data/")

# 1. 拆分每行数据，构造(复合key, 数值化分数)的RDD
data = lines.map(lambda x: x.split(',')) \
            .map(lambda parts: (
                # 复合key：(第一列单词, 最后一列文档编号)
                (parts[0], parts[-1]),
                # 将中间四列字符串转为float类型，作为求和的数值
                (float(parts[1]), float(parts[2]), float(parts[3]), float(parts[4]))
            ))

# 2. 对相同复合key的分数进行逐列求和
summed_data = data.reduceByKey(lambda a, b: (
    a[0] + b[0],
    a[1] + b[1],
    a[2] + b[2],
    a[3] + b[3]
))

# 3. 可选：将结果格式化为更易读的结构（单词, 文档编号, 分数1总和, 分数2总和, 分数3总和, 分数4总和）
formatted_data = summed_data.map(lambda x: (
    x[0][0], x[0][1], x[1][0], x[1][1], x[1][2], x[1][3]
))

# 打印处理结果
formatted_data.pprint()

# 启动流处理并等待终止（替换time.sleep，保证流持续运行直到手动停止）
ssc.start()
ssc.awaitTermination()

关键改进点说明： #

• 复合Key设置：用(parts[0], parts[-1])作为分组的key，确保只有单词和文档编号都完全匹配的行才会被聚合，这是实现需求的核心。
• 类型转换：拆分后的原始数据是字符串类型，必须转成float才能进行数值求和，否则会出现字符串拼接的错误（比如"-0.0"+"-0.494"会变成无效的字符串而非数值和）。
• 可靠的流运行方式：用ssc.awaitTermination()代替time.sleep(5)，前者会持续监听流数据直到手动停止，避免因为等待时间过短导致数据未处理完成就退出。
• 结果格式化：把复合key拆分开来，让输出结果更直观，方便你查看每个单词-文档组合的求和数据。

如果需要跨批次维护聚合状态（比如累计所有批次的求和结果），可以改用updateStateByKey，但如果只是每个批次内的独立求和，上面的reduceByKey就足够满足需求了。

内容的提问来源于stack exchange，提问作者Kush Singh