Spark DataFrame按条件拼接多列为新列的Scala优化实现问询

我有一个如下结构的Spark DataFrame，需要根据特定条件检查Field1到Field5列，生成名为EvaluatedReasons的新列。

示例DataFrame代码 #

import spark.implicits._
val df = Seq(
  ("C002", Some(1),Some("edf"),Some(10),Some(200),Some(500)),
  ("C001", Some(2), None,None, None,None) , // null场景
  ("C003", Some(1),Some("abc"),Some(80),Some(200),Some(900))   
).toDF("CustomerId", "Field1", "Field2", "Field3", "Field4", "Field5")

df.show(10,false)

预期输出 #

生成的EvaluatedReasons列需满足：

• C002：Field2 not equal to abc,Field5 < 800,
• C001：xyx, Field2 Is empty, Field3 Is empty, Field4 Is empty, Field5 Is empty,
• C003：Field3 < 10, Field4 < 100,

现有实现（冗余度高） #

我目前的实现可运行，但代码重复多，实际场景需检查8列，示例简化为5列：

val resultDf = (df
  .withColumn("Field1Reason", when(col("Field1") =!= "1", lit("xyx, ")).otherwise(lit("")))
  .withColumn("Field2Reason", when(col("Field2").isNull, lit("Field2 Is empty, ")).otherwise(lit("")))
  .withColumn("Field3Reason", when(col("Field3").isNull, lit("Field3 Is empty, ")).otherwise(lit("")))
  .withColumn("Field4Reason", when(col("Field4").isNull, lit("Field4 Is empty, ")).otherwise(lit("")))
  .withColumn("Field5Reason", when(col("Field5").isNull, lit("Field5 Is empty, ")).otherwise(lit("")))
  .withColumn("Field2NotNullReason", when(col("Field2").isNotNull && col("Field2") =!= "abc", 
                                           lit("Field2 not equal to abc,")).otherwise(lit("")))
  .withColumn("Field3NotNullReason", when(col("Field3").isNotNull && col("Field3") < 10, lit("Field3 < 10, ")).otherwise(lit("")))
  .withColumn("Field4NotNullReason", when(col("Field4").isNotNull && col("Field4") < 100, lit("Field4 < 100, ")).otherwise(lit("")))
  .withColumn("Field5NotNullReason", when(col("Field5").isNotNull && col("Field5") < 800, lit("Field5 < 800,")).otherwise(lit("")))
  .withColumn("EvaluatedReasons",
    concat(
      col("Field1Reason"), col("Field2Reason"), col("Field3Reason"),
      col("Field4Reason"), col("Field5Reason"), col("Field2NotNullReason"),
      col("Field3NotNullReason"), col("Field4NotNullReason"), col("Field5NotNullReason")
    )
              
  ).drop("Field1Reason", "Field2Reason", "Field3Reason", "Field4Reason", "Field5Reason",
        "Field2NotNullReason", "Field3NotNullReason", "Field4NotNullReason", "Field5NotNullReason")
)

更优雅的实现方案 #

通过集中定义检查规则的方式，避免重复代码，同时提升扩展性：

实现思路 #

1. 把每个字段的检查条件和对应提示文本封装成规则列表
2. 遍历规则生成对应的when表达式（满足条件返回提示，否则返回空字符串）
3. 拼接所有表达式生成EvaluatedReasons列

代码实现 #

import org.apache.spark.sql.functions.{col, lit, when, concat_ws}

// 定义所有检查规则：(条件表达式, 满足条件时的提示文本)
val checkRules = Seq(
  // Field1规则
  (col("Field1") =!= 1, "xyx, "),
  // Field2规则：空值检查 + 非空取值检查
  (col("Field2").isNull, "Field2 Is empty, "),
  (col("Field2").isNotNull && col("Field2") =!= "abc", "Field2 not equal to abc,"),
  // Field3规则
  (col("Field3").isNull, "Field3 Is empty, "),
  (col("Field3").isNotNull && col("Field3") < 10, "Field3 < 10, "),
  // Field4规则
  (col("Field4").isNull, "Field4 Is empty, "),
  (col("Field4").isNotNull && col("Field4") < 100, "Field4 < 100, "),
  // Field5规则
  (col("Field5").isNull, "Field5 Is empty, "),
  (col("Field5").isNotNull && col("Field5") < 800, "Field5 < 800,")
)

// 生成每个规则对应的表达式
val reasonExprs = checkRules.map { case (cond, msg) =>
  when(cond, lit(msg)).otherwise(lit(""))
}

// 拼接所有表达式生成目标列
val resultDf = df.withColumn("EvaluatedReasons", concat_ws("", reasonExprs: _*))

resultDf.show(10, false)

优势 #

• 扩展性强：新增字段检查时，只需在checkRules中添加规则，无需修改后续逻辑
• 代码简洁：避免大量重复的withColumn和drop操作
• 可读性高：所有规则集中定义，便于维护和修改

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