在处理海量数据时，JSON 格式由于其灵活的层级结构被广泛使用。然而，在进行数据分析或导入关系型数据库之前，通常需要将嵌套的 JSON 结构"压平"（Flatten）。Spark SQL 提供了强大的内置函数，特别是 explode 和点号选择符，能够高效地完成这一任务。

处理深度嵌套的数组与对象

假设我们有一段复杂的 JSON 数据，其中包含了嵌套数组以及多层对象结构。目标是将这些深层字段提取出来，形成扁平化的表格视图。

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

// 模拟复杂的嵌套数据
val rawJson = """{
  "batch_id": 5001,
  "category": "IoT_Sensor",
  "measurements": [
    {
      "sensor_id": "S1",
      "position": [["120.1", "30.2"]],
      "metadata": {
        "timestamp": "2023-10-01 10:00:00",
        "firmware": "v1.0.2"
      }
    },
    {
      "sensor_id": "S2",
      "position": [["120.5", "30.8"]],
      "metadata": {
        "timestamp": "2023-10-01 11:00:00",
        "firmware": "v1.0.5"
      }
    }
  ]
}"""

// 将字符串转为 DataFrame
val baseDf = spark.read.json(Seq(rawJson).toDS())

// 第一步：使用 explode 函数展开顶级数组
val explodedDf = baseDf.select(
  col("batch_id"),
  col("category"),
  explode(col("measurements")).as("obs")
)

// 第二步：通过点号和 getItem 获取深层属性
val finalDf = explodedDf.select(
  col("batch_id"),
  col("category"),
  col("obs.sensor_id").as("sid"),
  col("obs.position").getItem(0).getItem(0).as("longitude"),
  col("obs.position").getItem(0).getItem(1).as("latitude"),
  col("obs.metadata.timestamp").as("ts"),
  col("obs.metadata.firmware").as("ver")
)

finalDf.show()

执行上述逻辑后，原本嵌套在 measurements 数组中的数据会被拆分成多行，且 position 数组内部的坐标点和 metadata 对象中的字段都会被映射为独立的列。

处理时间序列或统计类数组

另一种常见的场景是 JSON 中包含了一个代表时间序列或特定维度的统计数组。我们需要将这些数组元素平铺开来，以便进行聚合分析。

// 模拟统计数据
val statsJson = """{
  "store_code": "CN_SH_001",
  "daily_performance": [
    {"score": 95.5, "period": 1},
    {"score": 92.0, "period": 7},
    {"score": 88.5, "period": 30}
  ]
}"""

val statsRdd = sc.parallelize(Seq(statsJson))
val statsDf = spark.read.json(statsRdd)

// 展开数组并提取内部字段
val flattenedStats = statsDf.select(
  col("store_code"),
  explode(col("daily_performance")).as("perf")
).select(
  col("store_code"),
  col("perf.score").as("metric_score"),
  col("perf.period").as("days_range")
)

flattenedStats.show()

在上述代码中，explode 函数将 daily_performance 数组里的每一个结构体元素转换成了一行数据。随后，通过 perf.score 这种引用方式，我们可以轻松访问结构体内部的属性并为其指定别名。

核心技术总结

• explode：这是最关键的函数，它接收一个数组（Array）或映射（Map）列作为输入，并为其中的每个元素生成一行。
• 点号引用 (Parent.Child)：对于 StructType 类型的列，直接使用 parent_column.child_field 即可访问其内部成员。
• getItem(index)：专门用于处理数组类型的列，通过索引获取指定位置的元素。