如何用PySpark/Pandas编写通用代码扁平化复杂嵌套JSON至DataFrame？

是否可行？ #

完全可行，但不存在绝对适用于所有场景的通用代码——因为JSON可能包含极端复杂的结构（比如循环引用、无限嵌套、异构数组等）。但针对绝大多数常见的嵌套场景（对象嵌套、数组嵌套），可以实现通用化程度极高的代码覆盖。

Pandas 通用扁平化实现 #

通过递归遍历JSON的嵌套层级，将嵌套键用点号连接，数组元素用索引标注，最终转为DataFrame：

import pandas as pd

def flatten_json(nested_json, prefix=""):
    flattened = {}
    for key, value in nested_json.items():
        new_key = f"{prefix}.{key}" if prefix else key
        if isinstance(value, dict):
            flattened.update(flatten_json(value, new_key))
        elif isinstance(value, list):
            for idx, item in enumerate(value):
                if isinstance(item, dict):
                    flattened.update(flatten_json(item, f"{new_key}[{idx}]"))
                else:
                    flattened[f"{new_key}[{idx}]"] = item
        else:
            flattened[new_key] = value
    return flattened

# 使用示例
nested_data = {
    "user": {"name": "Alice", "contacts": {"email": "alice@example.com", "phones": ["12345", "67890"]}},
    "order": {"id": 123, "items": [{"product": "Laptop", "price": 999}, {"product": "Mouse", "price": 25}]}
}
flattened_data = flatten_json(nested_data)
df = pd.DataFrame([flattened_data])
print(df)

注意：如果数组中存在大量异构元素（同时包含字典、字符串、数字），会生成较多空值，可根据业务需求调整数组的处理逻辑（比如将数组转为分隔字符串）。

PySpark 通用扁平化实现 #

利用PySpark的Schema解析能力，递归展开嵌套Struct类型，对Array类型先用explode转为多行再处理：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, explode
from pyspark.sql.types import StructType, ArrayType

def flatten_schema(schema, prefix=""):
    fields = []
    for field in schema.fields:
        field_name = f"{prefix}.{field.name}" if prefix else field.name
        if isinstance(field.dataType, StructType):
            fields.extend(flatten_schema(field.dataType, field_name))
        elif isinstance(field.dataType, ArrayType):
            # 展开数组为多行
            exploded_col = explode(col(field.name)).alias(f"{field_name}_exploded")
            fields.append(exploded_col)
            # 递归处理数组元素的嵌套结构
            elem_schema = field.dataType.elementType
            if isinstance(elem_schema, StructType):
                fields.extend(flatten_schema(elem_schema, f"{field_name}_exploded"))
        else:
            fields.append(col(field.name).alias(field_name))
    return fields

# 使用示例
spark = SparkSession.builder.appName("FlattenJSON").getOrCreate()
json_df = spark.read.json("path/to/your/nested.json")

# 生成扁平化字段列表并构建DataFrame
flattened_fields = flatten_schema(json_df.schema)
flattened_df = json_df.select(flattened_fields)
flattened_df.show()

注意：explode会将数组的每个元素转为一行，若需保留单行结构，可替换为concat_ws等函数将数组转为分隔字符串。

局限性说明 #

1. 无法处理循环引用的JSON（如A嵌套B，B又嵌套A），需额外添加循环检测逻辑。
2. 极端深度的嵌套可能引发性能问题，可通过限制递归深度或分批次处理优化。
3. 异构数组会生成大量空值，需结合业务场景做针对性清洗。

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