合并Delta Table时在Merge阶段实现源DataFrame去重的方法

首先明确你之前尝试无效的原因：

• 省略whenMatched逻辑无法解决重复问题：你示例中id=4的重复数据在目标Delta表中原本不存在，所有重复行都会命中whenNotMatched插入分支，自然会写入3条重复记录
• 添加whenMatchedDelete逻辑误删全表：本质是匹配条件设置错误，且该逻辑仅能处理目标表已存在的匹配行，完全无法覆盖新数据插入场景的重复问题。

Delta Merge本身的语义决定了：如果传入Merge的源数据集存在同匹配键的重复行，未匹配到目标表的重复行都会触发插入逻辑，不需要单独提前对待合并DataFrame做落地去重，直接在Merge定义中内联去重逻辑即可，属于Merge执行流程的一部分，实现方式如下：

基础实现（重复数据完全一致场景） #

如果同主键的重复数据内容完全一致，直接在Merge的源参数中调用去重算子即可，不需要额外预处理步骤，代码示例（PySpark API）：

from delta.tables import DeltaTable

# 加载目标Delta表
target_delta = DeltaTable.forName(spark, "your_target_delta_table")
# 待接入的带重复数据的DataFrame
source_df = spark.read.load("/path/to/new/data") # 替换成你的数据源加载逻辑

# 执行Merge，去重逻辑内联在源端定义中
target_delta.alias("t")\
    .merge(
        # 核心：按主键去重，该计算在Merge执行时完成，属于Merge流程的一部分
        source_df.dropDuplicates(["id"]).alias("s"),
        "t.id = s.id" # 基于主键做匹配关联
    )\
    .whenMatchedUpdateAll() # 匹配到主键存在则更新全字段，不需要更新可省略该行
    .whenNotMatchedInsertAll()\
    .execute()

执行后最终表效果和你预期完全一致，id=4的记录仅会保留1条。

进阶实现（重复数据存在差异场景） #

如果同主键的重复数据内容不一致，需要保留特定版本（比如最新时间的记录），可以在源端用窗口函数筛选目标记录后再传入Merge，示例：

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

# 定义窗口规则：按id分组，按更新时间倒序取第一条
id_window = Window.partitionBy("id").orderBy(F.desc("update_time"))
# 内联筛选每个主键的最新记录
deduplicated_source = source_df\
    .withColumn("row_num", F.row_number().over(id_window))\
    .filter(F.col("row_num") == 1)\
    .drop("row_num")

# 将deduplicated_source作为Merge的源传入即可，后续逻辑和上述基础实现一致

注意事项 #

• Delta 2.0及以上版本已经对Merge源端的去重逻辑做了执行计划优化，内联去重不会产生额外的性能开销，和提前单独对DataFrame去重的执行效率完全一致
• 不要尝试通过whenMatched相关分支处理源数据重复问题：该类分支仅对目标表已存在主键的记录生效，新插入数据的重复问题必须在源端传入Merge前（含内联在Merge参数中）完成去重。

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