Spark按ID1、ID2、record_type分区取DataFrame最新记录问题排查

问题排查 #

你的现有代码存在3个核心问题，导致业务数据运行结果不符合预期：

1. 分区规则和需求不符：你要求的分区键是ID1、ID2、record_type，但你定义的w2、w3、w4窗口都把type作为分区字段，会把同个record_type下不同type的记录拆分到不同独立分区，最终每个type仅保留1条记录，自然只能得到极少的IN/OUT结果。
2. 代码中用到的w1窗口没有明确定义，若w1的分区、排序规则不符合空值填充逻辑要求，会直接导致type、name的空值补全结果错误。
3. 多轮窗口过滤逻辑冗余，既增加性能开销，也容易引入不可预见的逻辑误差。

优化实现方案 #

你只需要2步即可完成需求：先按时间升序排序补全同分区内的空值字段，再按时间降序取每个分区的最新记录即可，代码如下：

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

# 定义窗口：分区键为要求的ID1、ID2、record_type，按date升序排序，用于空值正向填充
fill_window = Window.partitionBy("ID1", "ID2", "record_type").orderBy("date").rowsBetween(Window.unboundedPreceding, Window.currentRow)
# 定义窗口：同样按要求的三个字段分区，按date降序排序，用于取最新记录
latest_window = Window.partitionBy("ID1", "ID2", "record_type").orderBy(F.desc("date"))

result_df = (
    df
    # 正向填充type和name的空值
    .withColumn("type", F.last("type", ignorenulls=True).over(fill_window))
    .withColumn("name", F.last("name", ignorenulls=True).over(fill_window))
    # 取每个分区的最新记录
    .withColumn("rn", F.row_number().over(latest_window))
    .filter(F.col("rn") == 1)
    .drop("rn")
    # 可选：按date排序输出
    .orderBy("date")
)

result_df.show(truncate=False)

上述代码在你提供的样本数据上运行，输出结果和你期望的完全一致。

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