PySpark大数据分组去重聚合：如何替代低效的groupByKey？

最优方案：使用DataFrame内置聚合函数collect_set #

Spark的DataFrame API提供了专门的去重聚合函数collect_set，它会自动在分区内做预聚合，大幅减少shuffle阶段的数据传输量，性能远优于你当前使用的RDD groupByKey方案。代码实现如下：

from pyspark.sql import functions as F

# 按User、Date、Dept分组，聚合去重的Item列表
result_df = df.groupBy("User", "Date", "Dept") \
              .agg(F.collect_set("Item").alias("Unique_Items"))

# 查看结果
result_df.show(truncate=False)

这个方案的优势：

• 内置函数经过Spark优化，会自动执行局部预聚合，避免了groupByKey直接拉取全量数据的问题
• DataFrame采用Tungsten执行引擎和高效序列化机制，比RDD的性能更高
• 代码更简洁易读，维护成本低

若坚持用RDD：用reduceByKey实现预聚合 #

如果一定要基于RDD实现，可通过reduceByKey先在分区内合并去重集合，再全局聚合，避免groupByKey的低效问题。代码如下：

# 将每条记录映射为(分组键, 单元素集合)，再通过reduceByKey合并集合
rdd_result = df.rdd.map(lambda x: ((x.User, x.Date, x.Dept), {x.Item})) \
                   .reduceByKey(lambda a, b: a.union(b)) \
                   .mapValues(list) \
                   .toDF(["Group_Key", "Unique_Items"]) \
                   .selectExpr("Group_Key._1 as User", "Group_Key._2 as Date", "Group_Key._3 as Dept", "Unique_Items")

# 查看结果
rdd_result.show(truncate=False)

这里的核心是初始将Item转为单元素集合，reduceByKey会在每个分区内先合并相同分组键的集合，再将分区结果shuffle到全局合并，相比groupByKey减少了shuffle的数据量。

超大数据集的优化必要性 #

非常有必要优化：
当数据量达到GB/TB级时，groupByKey会将所有相同分组键的数据直接拉到同一节点，导致大量跨节点数据传输（shuffle），极易引发内存溢出或执行超时。而优化后的方案（collect_set或reduceByKey）通过预聚合减少了shuffle的数据量，同时利用Spark的优化机制提升执行效率，能显著缩短任务运行时间、降低资源消耗。

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