Delta表Merge操作报错验证：多源行匹配同一目标行的修正方案

报错原因 #

这个错误的核心是源表partdf中存在多条primary_key_hash相同的记录。Delta Lake的Merge操作遵循SQL语义，当多条源记录匹配到目标表（deltadf）的同一行时，系统无法确定使用哪条源数据来更新或插入目标行，因此抛出冲突错误。

去重代码的有效性 #

你编写的去重逻辑是可行的：

• 通过Window.partitionBy("primary_key_hash")按主键哈希分组
• 用row_number()按Id排序后筛选每组第一条记录
• 最终得到的df3中每个primary_key_hash仅保留一条记录，从根源上避免了多源行匹配同一目标行的问题

修改后的Merge代码验证 #

你将Merge语句中的partdf替换为去重后的df3是正确的，这样就能消除报错。不过注意代码中的一个小语法问题：whenNotMatchedInsertAll()后面多余的句号需要删除，否则会引发语法错误。修正后的完整代码如下：

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

# 先执行去重逻辑
df2 = partdf.withColumn("rn", row_number().over(Window.partitionBy("primary_key_hash").orderBy("Id")))
df3 = df2.filter("rn = 1").drop("rn")

# 执行合并操作
try:
  if allowDuplicates == "true":
    (deltadf.alias("t")
       .merge(
        df3.alias("s"),
        f"s.primary_key_hash = t.primary_key_hash")
      .whenNotMatchedInsertAll()
     .execute()
    )
  else:
    (deltadf.alias("t")
       .merge(
        df3.alias("s"),
        "s.primary_key_hash = t.primary_key_hash")
      .whenMatchedUpdateAll("s.change_key_hash != t.change_key_hash")
      .whenNotMatchedInsertAll()
     .execute()
    )

注：将原代码中的<>替换为!=是为了代码更易读，两者在SQL语义中是等价的

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