
本文介绍使用 PySpark 显式定义嵌套结构 Schema 并通过 selectExpr 提取深层字段,实现对含可选嵌套对象(如 enrollment、sports)的 JSON 数据的完整展平。
本文介绍使用 pyspark 显式定义嵌套结构 schema 并通过 `selectexpr` 提取深层字段,实现对含可选嵌套对象(如 `enrollment`、`sports`)的 json 数据的完整展平。
在 PySpark 中处理嵌套 JSON 时,仅靠 parallelize + json.dumps 无法真正解析嵌套结构,只会将子对象序列化为字符串或保留为 StructType 列,导致后续无法直接访问内部字段。要实现语义级展平(即把 enrollment.type 变成顶层列 type),关键在于两步:显式声明嵌套 Schema 和 使用点号路径语法提取字段。
首先,需为 JSON 数据明确定义 StructType Schema,尤其要嵌套声明 enrollment 和 sports 字段为 StructType,并为其子字段指定类型与空值策略。这确保 Spark 能正确推断缺失字段(如第一个记录无 enrollment)、兼容异构结构,并避免运行时类型错误:
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, BooleanType
schema = StructType([
StructField("student_id", IntegerType(), nullable=False),
StructField("room_id", StringType(), nullable=False),
StructField("enrolled", BooleanType(), nullable=True),
StructField("enrollment", StructType([
StructField("type", StringType(), nullable=True),
StructField("date", StringType(), nullable=True)
]), nullable=True),
StructField("sports", StructType([
StructField("team", StringType(), nullable=True),
StructField("position", StringType(), nullable=True)
]), nullable=True)
])接着,将 JSON 字符串(注意:需是单个合法 JSON 数组字符串)传入 spark.read.json(),并绑定该 Schema。为适配 json() API(它期望 RDD[String],每行一个 JSON 对象),我们用 parallelize([json_string]) 将整个数组作为单元素 RDD 传入:
json_data = '''[
{"student_id": 1234, "room_id": "abc", "enrolled": false},
{"student_id": 4321, "room_id": "def", "enrolled": true, "enrollment": {"type": "home", "date": "01-01-2020"}},
{"student_id": 678, "room_id": "htf", "sports": {"team": "hockey", "position": "forward"}}
]'''
df = spark.read.schema(schema).json(spark.sparkContext.parallelize([json_data]))最后,使用 selectExpr(或等价的 select(df["enrollment.type"].alias("type"), ...))按路径提取嵌套字段。Spark 会自动处理缺失嵌套对象——若某行无 enrollment 字段,则 enrollment.type 返回 null,无需额外 coalesce 或 when 判断:
flattened_df = df.selectExpr(
"student_id",
"room_id",
"enrolled",
"enrollment.type as type",
"enrollment.date as date",
"sports.team as team",
"sports.position as position"
)
flattened_df.show()✅ 注意事项:
-
spark.read.json()默认按行解析(每行一个 JSON 对象),若原始数据是 JSON 数组,必须整体包裹为单字符串并用parallelize([s])传入,否则会解析失败; - 显式 Schema 是健壮性的核心:它使 Spark 知晓
enrollment是结构体而非字符串,从而支持点号访问; - 所有嵌套字段名在
selectExpr中需用双引号包裹(因含点号),别名用as关键字; - 若嵌套层级更深(如
address.city.zipcode),写法同理:"address.city.zipcode as zipcode"。
通过这一模式,你可系统性地展平任意深度的可选嵌套 JSON,生成规整的宽表结构,为后续分析与机器学习任务奠定可靠数据基础。


















