Spark中Delta表满足特定条件的Upsert操作实现咨询

嘿，我瞅出你这段Merge代码的问题啦——你把「时间差超过14天」的判断放在了匹配（ON）子句里，虽然逻辑上能覆盖部分场景，但不够直观，也容易让后续维护的同学懵圈。咱们换个写法，把匹配逻辑和更新条件拆开，用单个Merge语句就能完美覆盖你所有的业务规则：

val dfUniqueException = DeltaTable.forPath(outputFolder)
dfUniqueException.as("existing")
 .merge(dfNewExceptions.as("new"), "new.ExceptionId = existing.ExceptionId")
 .whenMatched("new.LastUpdateTime > date_add(existing.LastUpdateTime, 14)")
   .updateAll()
 .whenNotMatched()
   .insertAll()
 .execute()

咱们来拆解下这段代码怎么对应你的三个业务条件：

• 规则1：插入新数据：当传入的ExceptionId在Delta表中完全不存在时，whenNotMatched分支会触发，直接插入整条新数据，完美符合要求。
• 规则2：更新时间差超14天的数据：当ExceptionId已经存在，并且新数据的LastUpdateTime比表中已有数据的时间晚14天以上时，whenMatched后面的条件会满足，执行全量更新操作。
• 规则3：不更新时间差不足14天的数据：如果ExceptionId存在，但新数据的更新时间和表中数据的时间差不到14天，whenMatched的条件不成立，不会执行任何更新，同时也不会触发插入（因为ID已经存在），完全符合“不操作”的要求。

如果后续你不想全量更新所有字段，只想更新特定的几个字段，还可以把updateAll()换成updateExpr来精准控制，比如：

.whenMatched("new.LastUpdateTime > date_add(existing.LastUpdateTime, 14)")
  .updateExpr(Map(
    "LastUpdateTime" -> "new.LastUpdateTime",
    "IsDisplayed" -> "new.IsDisplayed",
    "Message" -> "new.Message",
    "ExceptionType" -> "new.ExceptionType",
    "ExceptionMessage" -> "new.ExceptionMessage",
    "FullException" -> "new.FullException"
  ))

这样写法更灵活，也能避免不必要的字段更新~

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