PySpark如何按JSON的key分组并计算另一key的均值

你的代码报错来自两个逻辑错误：

• groupBy后传入mapValues的参数是同组所有记录构成的可迭代对象，不是单条字典，无法直接执行x['points']取值
• RDD原生的mean()方法是计算整个数值RDD的全局均值，不支持直接在分组后的键值对RDD上按组计算均值

方案1：修正原有groupBy逻辑（适合小批量数据） #

直接修正分组后的均值计算逻辑，对每组迭代器单独计算总和与数量的比值即可：

def points_by_hour(dataset):
    return dataset.groupBy(lambda x: x["time"]) \
        .mapValues(lambda group_records: sum(r["points"] for r in group_records) / len(list(group_records))) \
        .collect()

注意：groupBy会将同组所有数据全量加载到内存，数据规模大时容易出现内存溢出，不建议在大规模数据场景使用。

方案2：aggregateByKey实现（大规模数据推荐） #

避免全量拉取分组数据，通过分区内预聚合、分区间合并的逻辑计算，性能和内存表现更好：

def points_by_hour(dataset):
    # 转换结构为 (time, (points值, 计数1))
    kv_rdd = dataset.map(lambda x: (x["time"], (x["points"], 1)))
    # 聚合得到每个time对应的points总和、记录总数
    sum_count_rdd = kv_rdd.aggregateByKey(
        zeroValue=(0, 0),
        seqFunc=lambda acc, cur: (acc[0] + cur[0], acc[1] + cur[1]),
        combFunc=lambda p1, p2: (p1[0] + p2[0], p1[1] + p2[1])
    )
    # 计算均值并返回结果
    return sum_count_rdd.mapValues(lambda x: x[0] / x[1]).collect()

两种方案执行后都会返回符合预期的结果：

[(3, 4.0), (5, 1.0)]

如果需要返回整数格式的均值，将除法计算替换为整数除法//，或对结果做int()类型转换即可。

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