Autoloader去重失效：基于modifiedon保留唯一ID问题排查

我通过Dataverse将D365 CRM的数据加载到Azure Data Lake中，已配置Synapse Link的append only功能以保留历史数据。

使用Autoloader读取ADLS数据的代码如下：

df = (
    spark.readStream
    .option("delimiter", ",")
    .option("quote", '"')
    .option("mode", "permissive")
    .option("lineSep", "\r\n")
    .option("multiLine", "true")
    .format("cloudFiles")
    .option("cloudFiles.format", source_format)
    .option("cloudFiles.schemaLocation", checkpoint_directory)
    .option("header", "false")
    .option("escape", '"')
    .schema(schema)
    .load(data_source)
)

接下来尝试对df进行转换，基于最新的modifiedon字段仅保留唯一ID（无重复）：

from pyspark.sql.functions import max
df = df.dropDuplicates()
df = df.withWatermark("modifiedon", "1 day")
agg = df.groupBy("id","modifiedon").agg(max("modifiedon").alias("max_modifiedon"))
agg = agg.select("id", "modifiedon")
final =df.join(agg, on=["id", "modifiedon"], how="inner")

随后将数据加载到Delta表（Azure Databricks）：

final.writeStream.format("delta").option("checkpointLocation", checkpoint_directory).trigger(availableNow=True).start("abfss://bronze@prd.dfs.core.windows.net/D365/msdyn_workorder_autoloader_nodups")

查询表后仍能看到重复的ID，且未正确选取max(modifiedon)对应的记录，请问哪里出错了？

核心错误点 #

• 分组逻辑错误：你在groupBy中同时指定了id和modifiedon，会把每个(id, modifiedon)组合当成独立分组，根本无法计算每个id对应的最大modifiedon值，正确分组应仅按id聚合。
• Watermark误用：设置的1天watermark会丢弃modifiedon超过1天的旧数据，与你保留历史数据最新记录的需求冲突；且该场景下完全不需要watermark。
• dropDuplicates无效：未指定列的dropDuplicates()会对全列去重，但同一id不同modifiedon的记录属于不同行，这步操作起不到去重作用，反而浪费资源。
• Checkpoint残留状态：若之前运行过错误的流任务，checkpoint目录会残留错误处理状态，重新运行时会基于旧状态继续处理，导致结果错误，需先清理该目录再重新执行。

修正后的代码 #

推荐使用窗口函数实现每个id仅保留最新modifiedon记录的需求，流处理中这种方式更可靠：

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

# 定义窗口：按id分区，按modifiedon降序排列
window_spec = Window.partitionBy("id").orderBy(df["modifiedon"].desc())

# 为每个id的记录添加行号，取行号为1的（最新）记录
df_with_row_num = df.withColumn("row_num", row_number().over(window_spec))
final = df_with_row_num.filter(df_with_row_num["row_num"] == 1).drop("row_num")

# 写入Delta表
final.writeStream.format("delta")\
    .option("checkpointLocation", checkpoint_directory)\
    .trigger(availableNow=True)\
    .start("abfss://bronze@prd.dfs.core.windows.net/D365/msdyn_workorder_autoloader_nodups")

额外注意事项 #

• 若数据源是持续追加的新记录（包含旧id的更新版本），可使用Delta表的merge操作实现upsert（更新旧记录+插入新记录），更高效维护最新状态。
• 确保modifiedon字段是Spark可识别的时间类型（如Timestamp），否则排序和聚合会出错。

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