Scala Spark SQL：when条件中用循环表达式实现动态列差异判断

核心逻辑拆解 #

你的PySpark代码核心是：

1. 对除firstname、middlename、lastname外的所有列，生成判断规则：若两表对应列值不等则返回列名，否则返回空字符串
2. 按优先级生成status列：
   • df1.id为空 → 标记added
   • df2.id为空 → 标记deleted
   • 存在至少一个差异列 → 标记updated
   • 其余情况 → 标记unchanged

Scala 实现代码 #

import org.apache.spark.sql.functions.{when, lit, array_remove, size, array}

// 生成差异判断表达式集合
val excludeCols = Set("firstname", "middlename", "lastname")
val conditions_ = df1.columns
  .filter(col => !excludeCols.contains(col))
  .map(c => when(df1(c) =!= df2(c), lit(c)).otherwise(lit("")))

// 构建status列
val status = when(df1("id").isNull, lit("added"))
  .when(df2("id").isNull, lit("deleted"))
  .when(size(array_remove(array(conditions_: _*), lit(""))) > 0, lit("updated"))
  .otherwise(lit("unchanged"))

关键语法说明 #

• Scala用map替代PySpark的列表推导式，通过过滤列集合生成条件表达式
• 传递可变参数到array()时，用conditions_: _*将Seq转为可变参数（对应PySpark的*conditions_语法）
• Scala中Spark的不等于运算符是=!=，和PySpark的!=语义一致
• 空字符串需用lit("")显式生成，避免类型不匹配问题

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