Spark Scala Dataframe：动态对非空列应用自定义函数的实现需求

核心思路 #

利用Spark的coalesce函数提取指定列子集中的唯一非空值，再对该值应用自定义函数生成结果列。coalesce会返回参数列表中第一个非空值，正好匹配每行仅一列非空的场景，且实现完全独立于列数量。

步骤与代码示例 #

1. 定义自定义函数
   先根据业务需求定义UDF（用户自定义函数），示例逻辑可替换为你的实际函数f：

   import org.apache.spark.sql.functions.udf
   
   // 自定义函数f，示例：将字符串转为大写
   val f = udf((input: String) => input.toUpperCase)
   

2. 动态处理指定列子集
   用列表存储目标列名，无需硬编码，动态生成coalesce参数：

   import org.apache.spark.sql.functions.{col, coalesce}
   
   // 示例原DataFrame
   val df = spark.createDataFrame(Seq(
     (1, "text1", null, null),
     (2, null, "text2", null),
     (3, null, null, "text3")
   )).toDF("Id", "A", "B", "C")
   
   // 指定列子集（可替换为任意数量的列名）
   val targetCols = List("A", "B", "C")
   
   // 生成Result列：提取非空值后应用自定义函数
   val resultDf = df.withColumn("Result", f(coalesce(targetCols.map(col): _*)))
   

3. 输出结果
   执行resultDf.show()会得到目标结构：

   +---+-----+-----+-----+--------+
   | Id|    A|    B|    C|  Result|
   +---+-----+-----+-----+--------+
   |  1|text1| null| null|  TEXT1|
   |  2| null|text2| null|  TEXT2|
   |  3| null| null|text3|  TEXT3|
   +---+-----+-----+-----+--------+
   

关键说明 #

• 列类型要求：确保指定列子集的所有列数据类型一致，coalesce要求输入列类型匹配，题目场景下通常满足该条件。
• 扩展性：只需修改targetCols列表的内容，即可适配任意数量的列，核心逻辑无需改动。
• 高效性：coalesce是Spark内置的高效函数，比手动遍历列的方式性能更优。

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