Spark处理含动态命名子Schema的JSON文件问题

针对你遇到的动态顶层键JSON导入Spark时Schema不一致的问题，推荐两种解决方案（不建议转成重复"user"键的JSON——该格式不符合JSON规范，解析时会丢失重复键对应的数据，优先选择数组格式转换）：

方案一：预处理JSON文件为数组格式 #

用脚本提前将原JSON转换为数组结构，之后直接用Spark读取即可：

import json

# 读取原JSON文件
with open("your_input.json", "r") as infile:
    raw_data = json.load(infile)

# 提取顶层值组成数组
array_data = list(raw_data.values())

# 写入标准数组格式的JSON文件
with open("output_array.json", "w") as outfile:
    json.dump(array_data, outfile, indent=2)

处理完成后，直接用spark.read.json("output_array.json")读取即可，Schema完全统一。

方案二：Spark读取时直接转换（无需预处理文件） #

如果不想修改原文件，可以在Spark中先读取为文本，再解析处理：

Python版本 #

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 定义用户数据的固定Schema
user_schema = StructType([
    StructField("id", StringType()),
    StructField("age", IntegerType()),
    StructField("gender", StringType())
])

# 读取文本文件并完成转换
df = spark.read.text("your_input.json") \
    .select(F.from_json(F.col("value"), F.MapType(StringType(), user_schema)).alias("user_map")) \
    .select(F.explode(F.map_values(F.col("user_map"))).alias("user")) \
    .select("user.*")

# 查看结果
df.show()

Scala版本 #

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

// 定义用户数据的固定Schema
val userSchema = StructType(Seq(
    StructField("id", StringType),
    StructField("age", IntegerType),
    StructField("gender", StringType)
))

// 读取并转换数据
val df = spark.read.text("your_input.json")
    .select(from_json($"value", MapType(StringType, userSchema)).alias("userMap"))
    .select(explode(map_values($"userMap")).alias("user"))
    .select("user.*")

df.show()

核心逻辑说明 #

1. 先以文本形式读取JSON文件，避免Spark自动推断出混乱的动态Schema
2. 用from_json将文本解析为MapType，其中键是动态顶层ID，值是用户数据
3. 用map_values提取所有用户数据，再通过explode将数组展开为多行记录
4. 最后选择用户数据的字段，得到Schema统一的DataFrame

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