Spark Delta表复杂增删改Merge操作问题排查与解决

需求规则 #

• 当新数据GUID非空，且新数据的GUID+Value与Delta表中现有行匹配时，用新数据的WFID和Uid更新对应行；
• 当新数据GUID为空，且新数据的WFID与Delta表中现有行匹配时，删除Delta表中的匹配行；
• 其他所有情况，将新数据行插入Delta表。

数据示例 #

初始Delta表数据（df1） #

新数据（dfNew） #

预期结果 #

现有代码问题分析 #

原代码的核心问题在于Merge的匹配条件和分支逻辑设计错误：

1. 匹配条件仅覆盖了GUID+Value匹配的场景，未包含新数据GUID为空时匹配WFID的删除场景；
2. 空值判断使用了错误的语法（<> null），Spark中判断空值必须用is not null/is null；
3. whenMatched分支顺序不合理，导致部分匹配逻辑无法触发。

正确实现代码 #

以下是修正后的完整Scala代码，重点调整了Merge部分的逻辑：

import org.apache.spark.sql.SparkSession
import io.delta.tables.DeltaTable
import org.apache.spark.sql.functions.col

val spark = SparkSession.builder()
      .master("local[1]")
      .appName("Test1")
      .getOrCreate();

val df1 = Seq(
    (Some("GUID1"), Some("WFID1"),  None,               1),
    (None,          Some("WFID1"),  Some("Uid1"),       1)
).toDF("GUID", "WFID", "Uid", "Value")

val deltaTablePath = "file:///D:/Test1DeltaLake"
df1.write.format("delta").mode("overwrite").save(deltaTablePath)

val deltaTable = DeltaTable.forPath(spark, deltaTablePath)
println("Original Delta table")
deltaTable.toDF.show

val dfNew = Seq(
    (Some("GUID1"), Some("WFID1"),  Some("Uid1"), 1),
    (None,          Some("WFID1"),  Some("Uid1"), 1),
    (Some("GUID2"), None,           Some("Uid2"), 1),
).toDF("GUID", "WFID", "Uid", "Value")
println("New DF for update")
dfNew.show

// 修正后的Merge逻辑
deltaTable.as("existing")
    .merge(
        dfNew.as("new"),
        // 匹配条件：覆盖两种场景
        """
          |(new.GUID is not null AND new.GUID = existing.GUID AND new.Value = existing.Value)
          |OR
          |(new.GUID is null AND new.WFID is not null AND new.WFID = existing.WFID)
          """.stripMargin
    )
    // 先处理删除分支：新数据GUID为空且匹配WFID
    .whenMatched("new.GUID is null")
    .delete()
    // 再处理更新分支：新数据GUID非空且匹配GUID+Value
    .whenMatched("new.GUID is not null")
    .update(Map(
        "WFID" -> col("new.WFID"),
        "Uid" -> col("new.Uid")
    ))
    // 插入不匹配的新数据
    .whenNotMatched
    .insertAll
    .execute()
    
println("Delta table after update")
DeltaTable.forPath(spark, deltaTablePath).toDF.show

关键修改点 #

1. 扩展匹配条件：同时覆盖"GUID非空时匹配GUID+Value"和"GUID为空时匹配WFID"两种场景；
2. 修正空值判断语法：用is not null/is null替代错误的<> null；
3. 调整分支顺序：先处理删除逻辑，再处理更新逻辑，确保符合需求优先级；
4. 明确分支触发条件：每个whenMatched分支都添加了清晰的触发条件，避免逻辑冲突。

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