PySpark实现SCD2：清理DataFrame并调整日期顺序

针对你遇到的问题——合并连续相同hash的行并修正日期区间以符合SCD2规范，以下是可直接落地的解决方案：

核心思路 #

通过分组标记连续相同维度的行，再对每组进行聚合，将连续重复的行合并为单一记录，同时更新START_DATE和END_DATE为完整的时间区间，替代单纯过滤重复行的方式。

步骤与代码实现 #

1. 导入依赖函数 #

from pyspark.sql import Window
from pyspark.sql.functions import col, lag, when, sum, min, max, first

2. 定义窗口并标记连续分组 #

首先按ID分区、START_DATE排序，用lag函数对比当前行与前一行的hash，标记新分组的起始行，再通过累加标记值生成唯一的组ID：

# 定义窗口：按ID分区，按START_DATE升序排序
window_spec = Window.partitionBy("ID").orderBy("START_DATE")

# 添加分组标识：当当前hash与前一行不同时，标记为新组
df_with_groups = df.withColumn(
    "is_new_group",
    # 第一行没有前一行，默认标记为新组
    when(lag(col("hash")).over(window_spec).isNull(), 1)
    .otherwise(when(lag(col("hash")).over(window_spec) != col("hash"), 1).otherwise(0))
).withColumn(
    # 累加标记值，生成连续相同hash的组ID
    "group_id",
    sum(col("is_new_group")).over(window_spec.rangeBetween(Window.unboundedPreceding, 0))
)

3. 聚合分组合并日期区间 #

按ID和group_id聚合，提取每组的最早生效日期、最晚失效日期，保留维度属性：

scd2_merged_df = df_with_groups.groupBy("ID", "group_id", "hash")\
    .agg(
        min("START_DATE").alias("START_DATE"),  # 组内最早的生效日期
        max("END_DATE").alias("END_DATE"),      # 组内最晚的失效日期（下一个事件的开始日期）
        first("l_open_date").alias("l_open_date")  # 保留初始打开日期（假设同组内值一致）
    )\
    .drop("group_id")\
    .orderBy("ID", "START_DATE")

4. 可选：处理当前有效行的END_DATE #

如果原始数据中最后一行的END_DATE为null（表示当前仍有效），可替换为SCD2标准的最大日期：

final_scd2_df = scd2_merged_df.withColumn(
    "END_DATE",
    when(col("END_DATE").isNull(), "9999-12-31").otherwise(col("END_DATE"))
)

关键说明 #

• 你之前的方案仅过滤了重复行，但未合并日期区间，导致失效日期仍保留原始单行的数值，无法覆盖连续重复行的完整时间范围。
• 分组标记的方式能精准识别连续相同维度的行，聚合后得到的START_DATE和END_DATE才是符合SCD2规范的完整区间。
• 若l_open_date在同组内存在差异，可根据业务需求替换first为min/max等聚合函数。

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