PySpark实现按ID分区、按created_date排序取最后value值

你遇到的问题是原窗口函数的默认帧范围限制了last()的取值范围，导致无法获取整个分区的最后一个value。

问题原因 #

原代码中的窗口Window.partitionBy('id').orderBy('created_date')默认使用的窗口帧是RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW，也就是说last('value')只会取到当前行及之前的最后一个value，而非整个id分组内最晚created_date对应的value。

解决方案 #

方案1：修改窗口帧为整个分区 #

通过指定rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)，让窗口覆盖当前分区的所有行，这样last()就能取到整个分区的最后一个值：

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

w = Window.partitionBy('id').orderBy('created_date').rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
df = df.withColumn('last_value', F.last('value').over(w))

方案2：倒序排序后取第一个值 #

将created_date倒序排列，此时first('value')就等价于原顺序的最后一个值，同样可以确保覆盖整个分区：

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

w = Window.partitionBy('id').orderBy(F.desc('created_date'))
df = df.withColumn('last_value', F.first('value').over(w))

方案3：分组聚合后关联 #

先通过分组获取每个id的最新created_date对应的value，再关联回原表，这种方式逻辑更直观：

import pyspark.sql.functions as F

# 获取每个id的最新记录对应的value
latest_values = df.groupBy('id').agg(
    F.first('value', ignorenulls=True).over(
        Window.partitionBy('id').orderBy(F.desc('created_date'))
    ).alias('last_value')
).distinct()

# 关联回原数据集
df = df.join(latest_values, on='id', how='left')

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