如何将指定嵌套JSON转换为目标结构的Spark DataFrame？

完整解决方案代码 #

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

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

# 读取嵌套JSON文件（必须开启multiLine处理多行JSON）
raw_df = spark.read.option("multiLine", True).json(json_s3_path)

# 1. 将myschema结构体拆分为表名与表详情的键值对并展开
table_level_df = raw_df.select(explode(map_entries(col("myschema"))).alias("table_info"))

# 2. 提取表名、加载类型，再将fields结构体拆分为列名与类型的键值对并展开
column_level_df = table_level_df.select(
    col("table_info.key").alias("TABLE"),
    col("table_info.value.load_type").alias("LOAD_TYPE"),
    explode(map_entries(col("table_info.value.fields"))).alias("column_info")
)

# 3. 提取列名、转换类型为大写，同时过滤掉不需要的列（如isdeleted）
final_df = column_level_df.select(
    col("TABLE"),
    col("LOAD_TYPE"),
    col("column_info.key").alias("COLUMN"),
    upper(col("column_info.value")).alias("COLUMN_TYPE")
).filter(col("COLUMN") != "isdeleted")

# 查看结果
final_df.show(truncate=False)

步骤说明 #

1. 读取JSON：开启multiLine=True是关键，因为你的JSON是多行格式，Spark默认按单行解析会导致结构识别错误。
2. 拆分表层级：用map_entries将myschema这个嵌套结构体转换成(表名, 表详情)的键值对数组，再用explode将数组拆分为每行对应一个表的记录。
3. 拆分列层级：从表详情中提取load_type，再用同样的map_entries+explode将fields结构体拆分为(列名, 列类型)的键值对，实现完全扁平化。
4. 格式化与过滤：将列类型转换为大写以匹配你的期望结果，同时过滤掉不需要的isdeleted列（如果不需要过滤可移除filter语句）。

为什么之前的尝试未成功？ #

你之前的代码仅完成了JSON读取，没有对嵌套的结构体进行键值对转换和行展开操作，因此得到的是嵌套的DataFrame结构，而非目标要求的扁平化表格。

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