PySpark分组前排序是否安全？分组取首值能否保证时序正确性？

直接给结论：完全不安全，不能保证你想要的结果。

我之前踩过这个坑，当时也天真以为先排序再分组就能保留顺序，但Spark的分布式执行模型根本不按这个逻辑走。原因很简单：

• 你执行df.orderBy('ProductId', 'Date')之后，数据确实会按指定顺序排序，但这个排序是基于当前的分区分布的。
• 接下来的groupBy('ProductId')会触发shuffle操作——Spark需要把所有相同ProductId的数据拉到同一个分区里处理。而shuffle过程中，数据的顺序是完全不被保留的，它只关心把相同key的数据凑在一起，完全不管原来的排序。

所以你最后用func.first('Price')拿到的，只是每个shuffle后分区里的第一个Price值，和这个ProductId的最早Date完全没关系，纯粹是随机的（取决于shuffle的哈希分配逻辑）。

那正确的做法是什么？有两种靠谱的方式：

方式一：使用窗口函数（兼容所有Spark版本） #

先按ProductId分区，再按Date排序，然后取每个分区的第一行数据：

from pyspark.sql import Window
from pyspark.sql import functions as func

window_spec = Window.partitionBy('ProductId').orderBy('Date')
df_result = df.withColumn('first_price', func.first('Price').over(window_spec)) \
              .dropDuplicates(['ProductId']) \
              .select('ProductId', 'first_price')

方式二：使用min_by函数（Spark 3.0+推荐） #

Spark 3.0之后提供了min_by（和max_by）函数，可以直接指定按某个列取最值对应的另一列值，代码更简洁：

df_result = df.groupBy('ProductId').agg(func.min_by('Price', 'Date').alias('earliest_price'))

这两种方式都能稳定地拿到每个产品最早日期对应的价格，完全不用担心shuffle打乱顺序的问题。

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