Spark 从带类型标记的嵌套数组JSON中生成指定结构DataFrame的方法

核心思路 #

已知type字段的固定枚举值为foo、bar，无需动态扫描全表提取可能的取值，仅需将嵌套在数组内的元素拆解后，按类型映射为独立列即可。

具体操作步骤 #

• 第一步：读取原始JSON源文件，获得包含嵌套items数组的初始DataFrame
• 第二步：对items数组执行explode拆解，将数组内每个独立元素拆分为单独行，提取出元素内的type、value两个字段
• 第三步：针对每类已知的type值，通过条件判断匹配对应记录的value值，聚合后生成独立列

参考实现代码（Scala 版本） #

首先导入所需的Spark内置函数：

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

完整处理逻辑：

// 1. 读取原始JSON
val rawDF = spark.read.json("你的JSON文件存储路径")

// 2. 处理多记录场景：给每条原始根记录加临时唯一ID，避免炸开后归属错乱
val indexedDF = rawDF.withColumn("row_id", monotonically_increasing_id())

// 3. 炸开items数组，提取type和value字段
val explodedDF = indexedDF
  .select(col("row_id"), explode(col("items")).as("item"))
  .select(col("row_id"), col("item.type").as("type"), col("item.value").as("value"))

// 4. 按已知type映射生成目标列
val resultDF = explodedDF
  .groupBy(col("row_id"))
  .agg(
    max(when(col("type") === "foo", col("value"))).as("foo"),
    max(when(col("type") === "bar", col("value"))).as("bar")
  )
  .drop("row_id") // 移除临时辅助列

最终结果校验 #

执行resultDF.printSchema()即可得到目标结构：

root
 |-- bar: integer (nullable = true)
 |-- foo: integer (nullable = true)

实现说明 #

• 因为已经明确掌握所有type的枚举值，整个流程跳过了动态pivot算子必须的全表distinct扫描步骤，数据量越大性能优势越明显
• 选择max做聚合的逻辑很简单：同一条原始JSON记录里，每个type只会出现一次，当when条件匹配到对应type时，max会直接返回对应的value值，匹配不到的位置自动填充null，完全匹配目标结构的nullable属性

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