PySpark中orderBy()为何修改DataFrame聚合结果

前置公共逻辑 #

统计目标为从销售数据中提取纽约州最常共同售出的商品组合，公共处理步骤如下：

• 初始化SparkSession，读取指定路径下2019年parquet格式销售数据
• 从OrderDate字段提取小时值生成Hour列
• 筛选State = 'NY'的数据得到DataFrame sales_NY，该表包含OrderID、Product、Quantity、Price等共10个字段

两版实现与差异表现 #

版本1逻辑 #

1. 直接对sales_NY按OrderID、State分组，使用collect_list('Product')聚合生成商品列表列ProductList
2. 新增列计算ProductList长度，过滤仅含单个商品的订单
3. 按ProductList分组统计出现频次，降序排序取Top10结果

版本2逻辑 #

仅在分组聚合前增加orderBy('OrderID', 'Product')排序步骤，其余逻辑和版本1完全一致，但输出的Top10商品组合、对应计数和版本1存在明显差异。测试将排序替换为sort()方法后仍得到相同差异结果。

运行环境 #

• JupyterLab v3.4.2
• PySpark v3.0.1
• Java v15

原因结论 #

该现象不是PySpark Bug，核心是对collect_list行为和Spark执行逻辑的认知偏差：

1. collect_list本身不保证聚合后列表内元素的顺序。分布式计算场景下，同一个OrderID对应的商品行可能分布在多个Executor的不同分区，聚合时拉取各分区数据的顺序没有确定性约束，不做额外处理的话，同一个订单的商品在ProductList中的排列顺序完全随机。例如同是购买「A、B」两件商品的订单，可能一部分聚合得到['A','B']，另一部分得到['B','A']，Spark会将这两个列表识别为完全不同的分组键，原本应该合并统计的同购组合被拆分，最终计数自然失真。
2. 聚合前添加的orderBy/sort不会实际生效。Spark的Catalyst优化器在生成执行计划时，会判定分组聚合前的全局排序对collect_list的聚合结果无强制语义要求，会将该排序节点优化移除，因此加排序后的执行逻辑本质上和版本1没有区别，只是因为分布式任务执行时的数据拉取顺序随机性，导致两次运行的分组拆分结果不一致，最终输出的Top10出现差异。

正确实现方式 #

不要依赖全局排序保证列表内元素顺序，直接在聚合阶段固定列表元素顺序即可，推荐写法如下：

from pyspark.sql.functions import collect_list, sort_array, size, count

# 按订单聚合生成有序商品列表
order_products = sales_NY.groupBy("OrderID", "State") \
    .agg(sort_array(collect_list("Product")).alias("ProductList")) \
    .filter(size("ProductList") > 1)

# 统计同购组合频次取Top10
top_10_combinations = order_products.groupBy("ProductList") \
    .agg(count("*").alias("frequency")) \
    .orderBy("frequency", ascending=False) \
    .limit(10)

注意：不建议用全局sort/orderBy解决该问题，分布式场景下全局排序开销极大，且Shuffle过程中如果没有显式指定分区内排序，优化器随时可能调整执行顺序，无法稳定保证聚合列表的元素顺序。

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