Spark Scala多值列转JSON列及JSON字段提取报错问题咨询

• 你遇到的报错核心原因是to_json函数返回值为字符串类型，并非Struct结构，自然无法使用.语法提取嵌套字段
• 之前的转换逻辑存在逻辑倒置：你需要先把Vision这类列中逗号分隔的key=value格式字符串解析为Struct/Map结构，之后再按需转JSON格式，而非直接把未解析的原始字符串包成struct转JSON

以下为无需手动逐个指定列的批量处理方案：

步骤1：定义多值字符串解析UDF #

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.MapType

// UDF功能：将逗号分隔的key=value字符串解析为Map结构
val parseMultiValueStr = udf((str: String) => {
  if (str == null || str.trim.isEmpty) {
    Map.empty[String, String]
  } else {
    str.trim.stripSuffix(",").split(",")
      .map(_.trim.split("=", 2))
      .filter(_.length == 2)
      .map(arr => arr(0) -> arr(1))
      .toMap
  }
})

步骤2：批量转换所有目标列 #

场景1：需要直接提取字段，无需存储为JSON #

直接将列转换为Map结构即可直接取字段，注意包含特殊字符的键名需要用反引号包裹：

// 批量处理所有列，若存在不需要转换的列可自行过滤columns数组
val finalDf = df1.select(
  df1.columns.map(c => parseMultiValueStr(col(c)).as(c)): _*
)

// 提取Vision列的HDR:E字段示例
finalDf.select(col("Vision.`HDR:E`")).show()

场景2：需要存储为JSON字符串格式 #

转JSON后如需提取字段，用from_json临时解析即可：

// 批量转换为JSON列
val jsonDf = df1.select(
  df1.columns.map(c => to_json(parseMultiValueStr(col(c))).as(c)): _*
)

// 从JSON列提取字段示例
val jsonSchema = MapType(StringType, StringType)
jsonDf.select(
  from_json(col("Vision"), jsonSchema).getItem("HDR:E").as("hdr_status")
).show()

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