Scala中Spark DataFrame的JSON字符串列转结构化Schema并保留其他列

最优方案直接使用Spark内置的from_json函数处理，无需落盘、全内存计算、性能最高，可直接保留原有非JSON列。

实现步骤 #

1. 显式定义JSON对应的Schema
   生产环境推荐提前定义Schema，避免自动推断带来的性能损耗和类型不符合预期的问题，对应你给出的预期Schema写法如下：

import org.apache.spark.sql.types._

val jsonSchema = StructType(Seq(
  StructField("testdata1", StructType(Seq(
    StructField("dat2", ArrayType(StructType(Seq(
      StructField("id", StringType, nullable = true)
    )), containsNull = true), nullable = true)
  )), nullable = true),
  StructField("id", StringType, nullable = true),
  StructField("name", StringType, nullable = true)
))

2. 调用from_json解析列并保留原有字段

import org.apache.spark.sql.functions.from_json

// 假设你原来的DataFrame叫rawDF
val resultDF = rawDF
  .withColumn("json_parsed", from_json($"value", jsonSchema))
  .select(
    $"json_parsed.*", // 展开解析后的JSON所有顶层字段
    $"country",
    $"process"
  )

3. 验证Schema
   调用resultDF.printSchema()即可看到完全符合你预期的结构：

root
 |-- testdata1: struct (nullable = true)
 |    |-- dat2: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- id: string (nullable = true)
 |-- id: string (nullable = true)
 |-- name: string (nullable = true)
 |-- country: string (nullable = true)
 |-- process: string (nullable = true)

可选配置 #

如果你的JSON字符串存在格式不规范的场景，可以传入可选参数调整解析规则：

val options = Map("allowUnquotedFieldNames" -> "true", "allowSingleQuotes" -> "true")
val resultDF = rawDF
  .withColumn("json_parsed", from_json($"value", jsonSchema, options))
  .select($"json_parsed.*", $"country", $"process")

原有方案失效原因说明 #

• 第一种落表再用spark.read.json读取的方案，是把每整行数据当作一个JSON字符串解析，自然会丢失country、process两个非JSON列
• 第二种调用to_json的方案是反向操作：该函数的作用是把struct类型的列转为JSON字符串，和你要实现的JSON字符串转struct的需求完全相反，所以不会生效

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