PySpark如何将任意层级嵌套JSON拆解为多行多列的指定格式数据

核心思路 #

• 嵌套结构的level、tag、key均为动态值，因此不使用固定Struct定义JSON schema，采用三层嵌套MapType解析JSON字符串，无需提前知晓任意键名即可完成解析
• 逐层调用explode函数展开Map类型的键值对，每次提取对应层级的名称，最终转换为要求的扁平行结构

完整实现代码 #

from pyspark.sql import functions as F
from pyspark.sql import types as T

# 解析JSON列为嵌套Map结构
df_parsed = df.withColumn(
    "levels_map",
    F.from_json(
        F.col("Levels"),
        # 三层Map对应结构：{level: {tag: {key: value}}}
        T.MapType(
            T.StringType(),
            T.MapType(
                T.StringType(),
                T.MapType(T.StringType(), T.StringType())
            )
        )
    )
)

# 第一层展开：提取level
df_level = df_parsed.select(
    "name",
    F.explode("levels_map").alias("level", "tags_map")
)

# 第二层展开：提取tag
df_tag = df_level.select(
    "name",
    "level",
    F.explode("tags_map").alias("tag", "kv_map")
)

# 第三层展开：提取key和value，得到最终结果
df_final = df_tag.select(
    "name",
    "level",
    "tag",
    F.explode("kv_map").alias("key", "value")
)

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

扩展说明 #

• 若JSON中value包含非字符串类型，可将最内层StringType替换为VariantType（Spark 3.4及以上版本支持），保留原始数据类型
• 本方案原生基于Spark SQL函数实现，性能远高于自定义UDF，适配动态数量的level、tag、key场景

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