PySpark按主键分组保留同列最新有效值的问题排查

原代码逻辑存在3处核心错误，导致中间变更过、后续事件未提及的字段取值异常：

• 未指定窗口范围时，PySpark窗口默认范围为RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW，直接调用last(col)只会取到当前行的原始字段值。Spark读取JSON时会自动给事件中不存在的字段填充null，因此SORTORDER=5、6行的COLUMND原始值为null，直接取last会覆盖SORTORDER=4时写入的35000。
• 字段更新判断逻辑错误：原代码用全量聚合的变更列超集判断是否更新字段，没有校验当前事件的changed_cols是否真的包含该字段，不符合CDC事件 仅changed_cols列出的字段为本次变更值，其余字段值无意义 的核心语义。
• 值传递逻辑错误：原代码直接取原始字段的窗口last值，没有实现“当前事件未修改该列时，沿用上一次变更的有效值”的递推逻辑。

核心实现思路：按ID分区、SORTORDER正序逐行递推每个字段的最新有效值，仅当字段出现在当前事件的changed_cols列表中时才更新值，否则直接继承上一行已经计算好的有效值；最终取每个ID下SORTORDER最大的行输出，同时聚合该ID所有出现过的去重变更列生成COL_LIST。

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

# 读取CDC事件数据，替换为实际输入路径
json_df = spark.read.json(input_file_path)

# 分区排序窗口：按ID分区，SORTORDER正序，用于逐行递推字段值
window_asc = Window.partitionBy("ID").orderBy("SORTORDER")
# 分区排序窗口：按ID分区，SORTORDER倒序，用于取每个ID的最新事件行
window_desc = Window.partitionBy("ID").orderBy(F.desc("SORTORDER"))

# 配置需要处理的业务字段列表
target_columns = ["COLUMNA", "COLUMNB", "COLUMNC", "COLUMND"]

# 逐字段递推有效值：当前事件修改了该字段则取当前值，否则继承上一行的值
for col_name in target_columns:
    json_df = json_df.withColumn(
        col_name,
        F.when(
            F.array_contains(F.col("changed_cols"), col_name),
            F.col(col_name)
        ).otherwise(
            F.lag(col_name, 1).over(window_asc)
        )
    )

# 计算行排名、聚合全量去重变更列列表
result_df = json_df.withColumn(
    "rank",
    F.rank().over(window_desc)
).withColumn(
    "COL_LIST",
    F.array_distinct(F.flatten(F.collect_list("changed_cols").over(window_asc)))
)

# 过滤每个ID的最新行，输出指定字段
final_df = result_df.filter(F.col("rank") == 1).select(
    "ID", *target_columns, "COL_LIST"
)

final_df.show(truncate=False)

执行上述代码可得到完全符合预期的输出：

+---+-------+-------+-------+-------+----------------------------------------+
|ID |COLUMNA|COLUMNB|COLUMNC|COLUMND|COL_LIST                                |
+---+-------+-------+-------+-------+----------------------------------------+
|244|null   |user   |null   |35000  |[ID, COLUMNA, COLUMNC, COLUMNB, COLUMND]|
|245|200.0  |null   |CLOSE  |null   |[ID, COLUMNA, COLUMNB, COLUMNC]         |
+---+-------+-------+-------+-------+----------------------------------------+

• 递推过程使用处理后的字段值做lag取值，而非原始字段值，从根本上避免了后续事件自动填充的null覆盖历史有效值的问题
• 字段更新判断严格匹配CDC事件语义，仅当字段在当前事件的changed_cols中时才更新，其余场景直接继承历史值
• 变更列列表通过全窗口收集后去重生成，覆盖每个ID生命周期内所有出现过的变更字段

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