PySpark两个RDD按Code关联后按Name分组聚合问题求解

首先明确两个核心错误点：

• RDD join要求两个参与关联的RDD都必须是(键, 值)的二元组结构，你当前使用的rdd2是(Code, Year, Delay)三元组，未将非键字段打包为值部分，join时会默认取前两位作为键值对，直接丢弃Delay字段，这是关联结果缺少Delay的核心原因。
• Python3不支持在lambda参数列表中直接解包嵌套元组，lambda (keys, values):这种写法不符合Python3语法规范，会触发参数缺失报错。

步骤1：预处理rdd2为符合join要求的键值对结构 #

将rdd2的Code作为键，(Year, Delay)打包作为值：

rdd2_kv = rdd2.map(lambda x: (x[0], (x[1], x[2])))

步骤2：执行关联并打平结构 #

关联后返回的结构为(Code, (Name, (Year, Delay)))，通过位置索引解包避免语法错误：

joined = rdd1.join(rdd2_kv).map(lambda x: (x[0], x[1][0], x[1][1][0], x[1][1][1]))
# 打平后每行结构为：(Code, Name, Year, Delay)

步骤3：按Name分组做Delay聚合运算 #

使用aggregateByKey实现计数、平均值、最小值、最大值的一次性计算：

# 提取(Name, Delay)作为键值对后执行聚合
delay_agg = joined.map(lambda x: (x[1], x[3])) \
    .aggregateByKey(
        # 初始值：(计数, Delay总和, 最小值, 最大值)
        (0, 0.0, float('inf'), float('-inf')),
        # 分区内聚合逻辑
        lambda acc, val: (acc[0]+1, acc[1]+val, min(acc[2], val), max(acc[3], val)),
        # 分区间合并逻辑
        lambda acc1, acc2: (acc1[0]+acc2[0], acc1[1]+acc2[1], min(acc1[2], acc2[2]), max(acc1[3], acc2[3]))
    ) \
    # 计算平均值，输出结构：(Name, 计数, 平均值, 最小值, 最大值)
    .map(lambda x: (x[0], x[1][0], x[1][1]/x[1][0] if x[1][0] > 0 else 0, x[1][2], x[1][3]))

示例运行结果 #

基于你提供的测试数据，最终输出为：

[('name1', 2, 1.625, 1.25, 2.0), ('name2', 1, -1.0, -1.0, -1.0)]

内容的提问来源于stack exchange，提问作者Jenny Char