Delta表更新已有数据并插入新数据时Merge操作报错解决方案

问题根源 #

这个错误的核心是你的源DataFrame里有多条记录满足Merge的匹配条件，对应Delta表中的同一行，Spark没法确定用哪条源数据来更新目标行，违反了Merge的SQL语义。
结合你的场景，主要有两个问题：

1. 匹配条件写得不对：把要更新的Weight (KG)也放进了匹配条件里，导致真正需要更新的记录匹配不到目标行，反而可能插重复数据；如果源数据里有同一主键+相同Weight的重复行，直接就触发错误。
2. 源数据有重复：同一主键（name、city、Date）下有多条记录，哪怕匹配条件改对了，Merge时还是会报错。

具体解决步骤 #

1. 修正Merge的匹配条件 #

匹配条件只保留能唯一标识一条记录的主键：也就是col1（name）、col2（city）、JSON里的Date字段，把Weight (KG)去掉——因为这是要更新的字段，不能当匹配依据。

condition = """
    target.col1 = source.col1 
    AND target.col2 = source.col2 
    AND get_json_object(target.col4, '$.Date') = get_json_object(source.col4, '$.Date')
"""

（注：你之前的condition里写的name/city/value应该是笔误，要对应实际列名col1/col2/col4）

2. 先给源数据去重，保留最新记录 #

在执行Merge之前，必须确保每个主键组合下只有一条最新的记录。可以根据你的业务逻辑选保留规则：比如按col3（日期）降序取最新，或者按JSON里的Date排序取最新。
示例代码（按col3降序留最新）：

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

# 定义窗口：按主键分组，按col3倒序排
window_spec = Window.partitionBy(
    "col1", 
    "col2", 
    get_json_object("col4", "$.Date")
).orderBy(col("col3").desc())

# 去重，只留每个分组的第一条（最新）记录
df_deduplicated = df_content_transformed.withColumn(
    "row_num", 
    row_number().over(window_spec)
).filter(col("row_num") == 1).drop("row_num")

3. 执行修正后的Merge #

用去重后的源DataFrame跑Merge：

from delta.tables import DeltaTable

deltaTable = DeltaTable.forPath(spark, "abfss://silver@X.dfs.core.windows.net/db/db_name/table/")

deltaTable.alias("target").merge(
    df_deduplicated.alias("source"),
    condition
) \
.whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()

额外优化建议 #

• 提前解析JSON字段：把col4里的JSON拆成单独的列（比如date、weight_kg），这样不仅能简化Merge条件，还能提升性能，不用反复调用get_json_object。示例：

  from pyspark.sql.functions import from_json, col
  from pyspark.sql.types import StructType, StructField, StringType, DoubleType
  
  json_schema = StructType([
      StructField("Date", StringType()),
      StructField("Weight (KG)", DoubleType())
  ])
  
  df_parsed = df_content_transformed.withColumn(
      "col4_parsed", 
      from_json(col("col4"), json_schema)
  ).select(
      "col1", "col2", "col3",
      col("col4_parsed.Date").alias("date"),
      col("col4_parsed.Weight (KG)").alias("weight_kg")
  )
  

  之后Merge的匹配条件直接写target.date = source.date就行，更简单高效。
• 加主键约束：如果你的Delta表支持，可以给表加主键约束（ALTER TABLE ... ADD CONSTRAINT PRIMARY KEY (col1, col2, date)），从表结构层面保证主键唯一，避免后续再出现重复数据问题。

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