PySpark动态读取嵌套JSON及同名字段处理方案咨询

核心思路 #

• 递归遍历DataFrame的Schema，自动提取所有叶子节点的完整路径
• 基于完整路径构建select表达式，一次性展开所有嵌套结构
• 重名字段通过完整路径自然区分，无需手动处理冲突

代码实现 #

from pyspark.sql.types import StructType

def get_all_field_paths(schema, parent_path=""):
    field_paths = []
    for field in schema.fields:
        current_path = f"{parent_path}.{field.name}" if parent_path else field.name
        if isinstance(field.dataType, StructType):
            # 递归处理嵌套结构体
            field_paths.extend(get_all_field_paths(field.dataType, current_path))
        else:
            # 叶子节点，记录完整路径
            field_paths.append(current_path)
    return field_paths

# 读取原始JSON文件
df = spark.read.option("multiline", True).json(loc)

# 获取所有字段的完整路径
all_field_paths = get_all_field_paths(df.schema)

# 动态生成select表达式，用完整路径作为列名
df_flattened = df.selectExpr([f"`{path}` as `{path}`" for path in all_field_paths])

重名字段处理说明 #

对于像period_start_on这类重复出现的字段，最终会生成带完整路径的列名：

• data.accounts.last_accounts.period_start_on
• data.accounts.next_accounts.period_start_on
  完全避免了字段名冲突，同时保留了字段的层级关系。

可选优化：排除指定字段 #

如果需要跳过某些字段，可以在递归函数中添加过滤逻辑：

def get_all_field_paths(schema, parent_path="", exclude_fields=None):
    exclude_fields = exclude_fields or []
    field_paths = []
    for field in schema.fields:
        current_path = f"{parent_path}.{field.name}" if parent_path else field.name
        if current_path in exclude_fields:
            continue
        if isinstance(field.dataType, StructType):
            field_paths.extend(get_all_field_paths(field.dataType, current_path, exclude_fields))
        else:
            field_paths.append(current_path)
    return field_paths

# 示例：排除resource_uri字段
all_field_paths = get_all_field_paths(df.schema, exclude_fields=["resource_uri"])

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