Spark Delta Merge中如何在关联条件里使用变量实现分区裁剪？

问题场景 #

原有Delta Merge代码如下：

deltaTable.as("existing")
            .merge(dfNewData.as("new"), "new.SerialNumber = existing.SerialNumber and new.RSid = existing.RSid")
            .whenMatched()
            .update(Map("Id1" -> col("new.Id1"), /* 约20个字段 */))
            .whenNotMatched()
            .insertAll()
            .execute()

为实现分区裁剪提升性能，尝试在关联条件中加入SerialMin和SerialMax变量过滤，但如下写法报错：

deltaTable.as("existing")
            .merge(dfNewData.as("new"), "existing.SerialNumber >= lit(SerialMin) and existing.SerialNumber < lit(SerialMax) and new.SerialNumber = existing.SerialNumber and new.RSid = existing.RSid")
            .whenMatched()
            .update(Map("Id1" -> col("new.Id1"), /* 约20个字段 */))
            .whenNotMatched()
            .insertAll()
            .execute()

解决方案 #

方式1：使用Scala字符串插值嵌入变量 #

直接将Scala变量通过字符串插值注入merge条件字符串，变量值会直接替换到SQL表达式中：

val mergeCondition = 
  s"existing.SerialNumber >= $SerialMin and existing.SerialNumber < $SerialMax " +
  "and new.SerialNumber = existing.SerialNumber and new.RSid = existing.RSid"

deltaTable.as("existing")
  .merge(dfNewData.as("new"), mergeCondition)
  .whenMatched()
  .update(Map("Id1" -> col("new.Id1"), /* 约20个字段 */))
  .whenNotMatched()
  .insertAll()
  .execute()

方式2：使用Column表达式构建merge条件 #

用Spark的Column API拼接条件，通过lit()函数引入Scala变量，这种方式更安全，可避免SQL注入风险：

import org.apache.spark.sql.functions._

val mergeCondition = 
  col("existing.SerialNumber") >= lit(SerialMin) && 
  col("existing.SerialNumber") < lit(SerialMax) &&
  col("new.SerialNumber") === col("existing.SerialNumber") &&
  col("new.RSid") === col("existing.RSid")

deltaTable.as("existing")
  .merge(dfNewData.as("new"), mergeCondition)
  .whenMatched()
  .update(Map("Id1" -> col("new.Id1"), /* 约20个字段 */))
  .whenNotMatched()
  .insertAll()
  .execute()

补充优化：提前过滤新数据（可选） #

为进一步提升性能，可先对dfNewData按SerialNumber范围过滤，减少参与merge的数据量：

val filteredNewData = dfNewData.filter(col("SerialNumber") >= lit(SerialMin) && col("SerialNumber") < lit(SerialMax))

deltaTable.as("existing")
  .merge(filteredNewData.as("new"), mergeCondition)
  .whenMatched()
  .update(Map("Id1" -> col("new.Id1"), /* 约20个字段 */))
  .whenNotMatched()
  .insertAll()
  .execute()

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