如何将嵌套JSON格式数据扁平化转换为Spark DataFrame

• 优先手动定义Schema，禁用自动推断
  Spark默认的Schema推断逻辑需要扫描样本数据，若嵌套层级深、部分样本products字段为空或格式不一致，会出现推断错误导致返回null。你需要按照实际JSON结构逐层定义Schema，示例如下：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, ArrayType

# 定义最内层products结构体的Schema
products_schema = StructType([
    StructField("product_id", StringType(), nullable=True),
    StructField("product_name", StringType(), nullable=True),
    StructField("price", IntegerType(), nullable=True),
    StructField("quantity", IntegerType(), nullable=True)
])

# 定义完整JSON的外层Schema
full_schema = StructType([
    StructField("order_id", StringType(), nullable=True),
    StructField("user_id", StringType(), nullable=True),
    StructField("order_time", StringType(), nullable=True),
    # 若products为单个结构体直接用products_schema，若为数组则套ArrayType
    StructField("products", products_schema, nullable=True)
    # 数组写法：StructField("products", ArrayType(products_schema), nullable=True)
])

• 检查读取参数配置
  如果你的JSON是跨多行存储、或整个文件为单个JSON数组，必须开启multiline参数，否则Spark会按单行解析导致结构读取不全：

df = spark.read \
    .option("multiline", "true") \
    .schema(full_schema) \
    .json("your_json_file_path")

如果存在字段名大小写不匹配的情况，可额外添加配置.option("caseSensitive", "false")关闭大小写敏感校验。

• 排查数据格式问题
  如果配置正确仍返回null，可先读取原始文本确认JSON结构是否一致：

# 先读取整行文本查看原始JSON内容
raw_df = spark.read.text("your_json_file_path")
raw_df.show(truncate=False)

同时可开启mode="PERMISSIVE"查看解析错误记录：

df = spark.read \
    .option("multiline", "true") \
    .option("mode", "PERMISSIVE") \
    .schema(full_schema) \
    .json("your_json_file_path")
# 查看解析失败的记录
df.filter("_corrupt_record is not null").show()

• 数组类型的嵌套结构体可展开后查询
  如果products为结构体数组，可使用explode函数展开后读取字段：

from pyspark.sql.functions import explode

df_exploded = df.withColumn("product_item", explode("products"))
df_exploded.select("order_id", "product_item.*").show(truncate=False)

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