如何在PySpark DataFrame中提取并扁平化字符串格式的嵌套JSON列

1. 将字符串列解析为结构化JSON #

首先需要把存储JSON字符串的列转换成PySpark可识别的结构化数据，核心用from_json函数，需要对应JSON结构的Schema（可自动推断或手动定义）。

• 导入依赖函数：

from pyspark.sql.functions import from_json, col

• 自动推断Schema（适合快速验证，生产环境建议手动定义）：

# 取一条JSON字符串样本生成Schema
sample_json = df.select("你的JSON列名").head()[0]
json_schema = spark.read.json(spark.sparkContext.parallelize([sample_json])).schema

# 解析JSON列，生成结构化列
df_parsed = df.withColumn("parsed_json", from_json(col("你的JSON列名"), json_schema))

• 手动定义Schema（适合结构固定的场景，更稳定）：
  比如假设嵌套JSON结构包含顶层id、嵌套对象details（含name和数组tags），Schema定义如下：

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

json_schema = StructType([
    StructField("id", IntegerType(), nullable=True),
    StructField("details", StructType([
        StructField("name", StringType(), nullable=True),
        StructField("tags", ArrayType(StringType()), nullable=True)
    ]), nullable=True)
])

df_parsed = df.withColumn("parsed_json", from_json(col("你的JSON列名"), json_schema))

2. 扁平化嵌套结构 #

根据嵌套层级和数据类型（对象/数组），选择对应的展开方式：

• 直接展开嵌套对象字段：
  如果是单层嵌套对象，可通过点语法直接提取字段：

df_flattened = df_parsed.select(
    # 保留原DataFrame的其他列
    col("原列1"),
    col("原列2"),
    # 提取结构化JSON中的字段并别名
    col("parsed_json.id").alias("json_id"),
    col("parsed_json.details.name").alias("user_name"),
    col("parsed_json.details.tags").alias("user_tags")
)

如果顶层嵌套字段较多，也可以用select("*", "parsed_json.*").drop("parsed_json")快速展开顶层所有字段。

• 处理数组类型字段：
  如果JSON中包含数组，用explode函数将数组的每个元素拆分为单独一行：

from pyspark.sql.functions import explode

# 先展开对象，再拆分数组
df_exploded = df_flattened.withColumn("single_tag", explode(col("user_tags"))).drop("user_tags")

3. 验证结果 #

用printSchema()查看解析后的结构，用show()预览数据，确认扁平化是否符合预期：

df_exploded.printSchema()
df_exploded.show(truncate=False)

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