如何用PySpark动态扁平化JSON字符串Payload至DataFrame列

问题说明 #

• 需求：将JSON字符串类型的payload完全拆分为独立列加载到PySpark DataFrame，且payload结构动态变化，无法提前固定Schema
• 数据情况：源文件为多行JSON，仅保留payload内的扁平化字段，移除外层所有列
• 当前问题：现有代码将payload处理为键值对Map，未拆分为独立列，无法直接用于后续分析

解决办法 #

核心思路 #

1. 解析payload字符串为JSON对象，避免硬编码Schema适配动态结构
2. 递归扁平化嵌套字典与数组，生成带点分隔符的扁平字段名（如detail.idx、transaction.data2.items.0.quantity）
3. 动态提取所有扁平字段，将Map类型结果展开为独立列

完整代码 #

import json
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, explode, map_keys, col
from pyspark.sql.types import MapType, StringType

def flatten_json(y):
    out = {}
    def flatten(x, name=''):
        # 处理嵌套字典
        if isinstance(x, dict):
            for key in x:
                flatten(x[key], name + key + '.')
        # 处理数组，按索引生成字段后缀
        elif isinstance(x, list):
            for idx, item in enumerate(x):
                flatten(item, name + str(idx) + '.')
        # 基础类型直接存入结果，空值保留为None
        else:
            out[name[:-1]] = str(x) if x is not None else None
    flatten(y)
    return out

# 初始化Spark会话
spark = SparkSession.builder \
    .appName("DynamicPayloadFlattener") \
    .getOrCreate()

# 读取多行JSON文件
df = spark.read.json("你的JSON文件路径")

# 提取payload并扁平化处理成Map结构
flatten_udf = udf(lambda json_str: flatten_json(json.loads(json_str)), MapType(StringType(), StringType()))
df_flattened_map = df.select(flatten_udf("payload").alias("flattened_payload"))

# 动态获取所有扁平后的字段名
all_fields = df_flattened_map.select(explode(map_keys(col("flattened_payload")))) \
                             .distinct() \
                             .rdd.flatMap(lambda row: row) \
                             .collect()

# 将Map展开为独立列
final_df = df_flattened_map.select(*[col("flattened_payload").getItem(field).alias(field) for field in all_fields])

# 查看最终扁平化结果
final_df.show(truncate=False)

关键步骤解释 #

• flatten_json递归函数：拆解嵌套JSON结构，数组元素会带上索引后缀（如cla.0、items.0.price），保证所有层级都被扁平化为一维键值对
• 动态字段提取：通过map_keys获取所有扁平字段名，无需提前知晓payload的具体结构，适配动态变化的场景
• 展开Map列：遍历所有字段，用getItem将每个键对应的值转为独立列，实现完全扁平化的效果

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