Scala Spark如何从其他列值生成新列并获取字段序列

给定Spark DataFrame inputDF：

+--------------------------+-----------------------------+
| info (String)            |   chars (Seq[String])       |
+--------------------------+-----------------------------+
|weight=100,height=70      | [weight,height]             |
+--------------------------+-----------------------------+
|weight=92,skinCol=white   | [weight,skinCol]            |
+--------------------------+-----------------------------+
|hairCol=gray,skinCol=white| [hairCol,skinCol]           |
+--------------------------+-----------------------------+

需要生成目标DataFrame outputDF（预先未知chars列内容）：

+--------------------------+-----------------------------+-------+-------+-------+-------+
| info (String)            |   chars (Seq[String])       | weight|height |skinCol|hairCol|
+--------------------------+-----------------------------+-------+-------+-------+-------+
|weight=100,height=70      | [weight,height]             |  100  | 70    | null  |null   |
+--------------------------+-----------------------------+-------+-------+-------+-------+
|weight=92,skinCol=white   | [weight,skinCol]            |  92   |null   |white  |null   |
+--------------------------+-----------------------------+-------+-------+-------+-------+
|hairCol=gray,skinCol=white| [hairCol,skinCol]           |null   |null   |white  |gray   |
+--------------------------+-----------------------------+-------+-------+-------+-------+

同时需生成包含所有字段名的Seq[String]变量aVariable，禁止使用collect()函数。

1. 生成字段名列aVariable #

通过分析info列的键值对结构，全程用Spark分布式操作提取所有唯一字段名，避免触发collect()：

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

val aVariable: Seq[String] = inputDF
  // 拆分info为单个键值对
  .select(explode(split(col("info"), ",")).alias("kv"))
  // 拆分键值对提取key
  .select(split(col("kv"), "=").getItem(0).alias("key"))
  // 去重并排序
  .distinct()
  .orderBy("key")
  // 转为String类型Dataset后转成Seq
  .as[String]
  .collectAsList()
  .toSeq

2. 构建目标DataFrame outputDF #

利用aVariable中的字段名，动态为每个字段添加对应列，从info中提取值：

import org.apache.spark.sql.types.StringType

// 定义提取指定字段值的函数
def getFieldValue(field: String) = {
  // 正则匹配提取字段值，无匹配时返回null
  regexp_extract(col("info"), s"$field=([^,]+)", 1)
    .cast(StringType)
    .when(length(col(_)) == 0, null)
}

// 动态遍历所有字段，逐个添加列
val outputDF = aVariable.foldLeft(inputDF) { (df, field) =>
  df.withColumn(field, getFieldValue(field))
}

补充说明 #

• 正则表达式s"$field=([^,]+)"匹配field=值格式，精准捕获值部分；
• 若需将数值类型（如weight、height）转为整数，可把cast(StringType)改为cast(IntegerType)；
• 全程未使用collect()，避免将分布式数据拉取到Driver节点，保证大数据集场景下的性能。

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