如何使用Pyspark将多层嵌套动态Key的JSON解析为多行数据

你提供的JSON中sensors和下属的work字段均为嵌套Map结构，只需两次explode对应Map即可打平为你需要的多行结构，无需使用arrays_zip。

完整代码示例 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, col

# 初始化SparkSession
spark = SparkSession.builder.appName("ParseSensorJson").getOrCreate()

# 读取JSON文件，若为单文件多行JSON需开启multiLine配置
df = spark.read.option("multiLine", "true").json("你的JSON文件路径.json")

# 第一层explode：打平sensors Map，得到传感器ID和对应属性
sensor_df = df.select(explode(col("sensors")).alias("sensor_id", "sensor_detail"))

# 提取传感器基础属性
sensor_base_df = sensor_df.select(
    col("sensor_id"),
    col("sensor_detail.fieldname").alias("fieldname"),
    col("sensor_detail.info").alias("info"),
    col("sensor_detail.id").alias("sensor_id_val"),
    col("sensor_detail.work").alias("work_map")
)

# 第二层explode：打平每个传感器下的work Map
work_explode_df = sensor_base_df.select(
    "*",
    explode(col("work_map")).alias("work_ts_key", "work_detail")
)

# 提取最终需要的所有字段，按你要求的顺序排列
final_df = work_explode_df.select(
    col("sensor_id"),
    col("fieldname"),
    col("info"),
    col("sensor_id_val"),
    col("work_detail.ts").alias("ts"),
    col("work_detail.te").alias("te"),
    col("work_detail.lat").alias("lat"),
    col("work_detail.lon").alias("lon")
)

# 输出为无表头的CSV格式
final_df.write.option("header", "false").csv("你的输出路径")

步骤说明 #

• 读取JSON时如果是单文件存储整个JSON结构，必须开启multiLine=true，否则会按行解析导致结构报错
• 两次explode分别对应两层嵌套Map，天然就能生成你需要的多行结构，不需要做额外的数组转换操作
• 你示例输出中部分行的字段拼接、数值重复属于手动输入笔误，上述代码输出的字段拆分和数值都是完全正确的

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