如何高效加载处理含异构演化Schema的JSON文件并适配Spark架构

你当前方案的核心瓶颈是spark.read.json(filtered_df.rdd.map(...))会触发全量数据从Executor拉取到驱动节点做Schema推断，而Schema推断实际只需要少量样本即可完成，只需做两处修改即可解决问题：

1. 提前缓存提取完表名的中间DataFrame，避免循环过滤时重复读取源数据
2. 推断Schema时仅使用少量采样数据，不需要全量数据参与

import pyspark.sql.functions as F

df = spark.read.text(directory)
with_table_df = (
    df
    .withColumn("table", F.get_json_object('value', '$.payload.table'))
    .withColumn("json_payload_data", F.get_json_object('value', '$.payload.data'))
).cache() # 新增：缓存中间结果，避免重复计算

unique_tables = with_table_df.select('table').distinct().rdd.map(lambda r: r[0]).collect()

for table in unique_tables:
    filtered_df = with_table_df.filter(f"table = '{table}'")
    # 修改：仅取1000条样本推断Schema，大幅减少拉取到驱动的数据量
    sample_rdd = filtered_df.select("json_payload_data").limit(1000).rdd.map(lambda row: row.json_payload_data)
    table_schema = spark.read.json(sample_rdd).schema

    changes_df = (
        filtered_df
        .withColumn('payload_data', F.from_json('json_payload_data', table_schema))
        .select('payload_data.*')
    )

    # 执行数据校验
    if valid:
        changes_df.write.mode("append").option("mergeSchema", "true").saveAsTable(target_table)

with_table_df.unpersist()

如果担心采样遗漏新增字段，可以把limit(1000)替换为sample(0.05)按比例采样，只要样本覆盖所有字段即可。

如果表数量多、单表数据量极大，可以采用全Executor侧计算的方案，全程只有Schema元数据返回驱动：

1. 先对每个表批量采样少量数据，生成表名到Schema的映射字典
2. 用CASE WHEN表达式匹配对应Schema，直接在Executor端完成所有数据的JSON解析

import pyspark.sql.functions as F

df = spark.read.text(directory)
with_table_df = (
    df
    .withColumn("table", F.get_json_object('value', '$.payload.table'))
    .withColumn("json_payload_data", F.get_json_object('value', '$.payload.data'))
)

# 每个表采样100条数据用于推断Schema，仅返回元数据到驱动
sample_df = with_table_df.groupBy("table").agg(F.slice(F.collect_list("json_payload_data"), 1, 100).alias("samples"))
table_schema_map = {}
for row in sample_df.collect():
    table_name = row["table"]
    sample_rdd = spark.sparkContext.parallelize(row["samples"])
    table_schema = spark.read.json(sample_rdd).schema
    table_schema_map[table_name] = table_schema.json() # 序列化为JSON字符串方便表达式调用

# 构造解析表达式，按表匹配对应Schema，全Executor端执行
when_clauses = "\n".join([f"WHEN '{t}' THEN from_json(json_payload_data, '{s}')" for t, s in table_schema_map.items()])
parse_expr = F.expr(f"CASE table {when_clauses} END AS payload_data")
parsed_df = with_table_df.withColumn("payload_data", parse_expr)

# 后续按表分组做校验、写入即可，全程数据不需要返回驱动

你提到的按表分区存储是长期运维成本最低、性能最好的方案：

• 上游导出JSON时按table字段做分区，每个表的数据写入独立的table=xxx子目录
• 消费时直接遍历分区目录，针对每个表目录调用spark.read.json即可，天然支持Schema推断和动态演化，不需要额外做字段提取、过滤操作
• 可直接对接Delta Lake、Iceberg等事务存储格式，自动处理Schema合并、ACID写入，整体性能比现有方案提升3~10倍

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