如何使用Pyspark将多JSON schema字段值追加到DataFrame对应列

值覆盖、无法生成多行结果的核心原因有三点：

• 未对不同schema的源文件做单独处理：不同结构的JSON文件如果直接批量读取，Spark会按合并后的schema做解析，后续withColumn是全量DF操作，后执行的同名字段赋值会直接覆盖之前的计算结果
• 字段映射逻辑错误：原代码将三个不同来源、本应对应Column A/B/C的字段，全部赋值给了Column A，本身就存在列名写错的问题
• 原has_column函数对嵌套结构字段判断失效：嵌套字段不存在时直接通过df[col]访问会触发Spark分析异常，无法正确返回布尔值，导致分支判断不符合预期

按照「单文件单独处理提取字段 -> 同结构DF合并」的逻辑改造即可，步骤如下：

• 替换原有不可靠的嵌套字段判断函数，支持按路径逐层校验嵌套字段是否存在
• 遍历所有待处理JSON文件，逐个读取生成独立的DataFrame，不提前合并不同schema的文件
• 对每个单文件DF，按照映射规则提取三个目标列：字段存在则取值，不存在则填充空字符串，保证所有单文件处理后列结构完全一致
• 用unionByName合并所有处理后的DF，即可得到每个文件对应一行的最终结果表

from pyspark.sql.functions import lit, when, col
from functools import reduce

def has_nested_column(df, col_path: str) -> bool:
    """校验嵌套字段是否存在，支持a.b.c格式的路径输入"""
    path_layers = col_path.split(".")
    current_schema_fields = df.schema.fields
    for layer in path_layers:
        match_field = None
        for field in current_schema_fields:
            if field.name == layer:
                match_field = field
                break
        if not match_field:
            return False
        # 进入下一层嵌套结构校验
        current_schema_fields = match_field.dataType.fields if hasattr(match_field.dataType, "fields") else []
    return True

# 替换为实际的JSON文件路径列表
source_file_paths = [
    "/path/to/first_pull_schema_file.json",
    "/path/to/second_cone_schema_file.json",
    "/path/to/third_var_schema_file.json"
]

# 定义字段映射关系：(目标输出列名, 源文件嵌套字段路径)
column_mapping_rules = [
    ("Column A", "sample_column.pull.notify.roid.alert"),
    ("Column B", "sample_column.cone.pull.notify.roid.title"),
    ("Column C", "sample_column.var.pull.notify.roid.action.pan.content")
]

processed_df_list = []
for file_path in source_file_paths:
    # 单文件单独读取，保留自身原始schema
    current_df = spark.read.json(file_path)
    # 按映射规则生成所有目标列
    for target_col_name, source_col_path in column_mapping_rules:
        if has_nested_column(current_df, source_col_path):
            current_df = current_df.withColumn(
                target_col_name,
                when(col(source_col_path).isNotNull(), col(source_col_path))
                .otherwise(lit("").cast("string"))
            )
        else:
            current_df = current_df.withColumn(target_col_name, lit("").cast("string"))
    # 仅保留目标输出列，保证后续union的列结构一致
    current_df = current_df.select([rule[0] for rule in column_mapping_rules])
    processed_df_list.append(current_df)

# 合并所有处理完成的同结构DataFrame
final_result_df = reduce(lambda df1, df2: df1.unionByName(df2), processed_df_list)

# 查看结果
final_result_df.show()

• 单文件处理阶段不会出现跨文件的字段值覆盖，每个文件只会提取自身存在的字段值，其余目标列自动填充空串
• 合并后的结果表共3行数据，分别对应3个源文件的提取结果，无值丢失、覆盖问题
• 嵌套字段判断逻辑可靠，不会因为schema差异触发Spark运行时异常

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