Spark读取带过滤的JSON文件：如何同步过滤Schema结构？

方法一：基于过滤后的数据重新推断Schema #

适合需要完全自动推断的场景，原理是将过滤后的数据转为JSON格式，让Spark重新读取并推断仅适用于当前数据集的Schema：

# 将过滤后的DataFrame转为JSON格式的RDD
filtered_json_rdd = filtered_df.toJSON()
# 重新读取JSON，Spark会自动基于当前数据推断Schema
trimmed_schema_df = spark.read.json(filtered_json_rdd)

如果过滤后的数据量较大，可以先采样减少开销：

# 采样10%的数据用于推断Schema（可根据实际调整比例）
sampled_rdd = filtered_df.sample(withReplacement=False, fraction=0.1).toJSON()
# 先推断Schema
trimmed_schema = spark.read.json(sampled_rdd).schema
# 用推断出的Schema重新处理全量过滤数据
trimmed_schema_df = spark.read.schema(trimmed_schema).json(filtered_json_rdd)

方法二：动态提取嵌套字段并重构Schema #

针对你的场景（仅event嵌套字段需要精简），可以精准提取过滤后数据中event实际存在的字段，重构列来同步Schema：

步骤1：获取event字段的所有实际存在的键 #

如果过滤后的数据中event结构统一，直接取一行样本即可：

# 获取第一行非空的event数据
sample_event = filtered_df.filter(col("event").isNotNull()).select("event").first()[0]
# 提取所有字段名
event_field_names = sample_event.asDict().keys()

如果过滤后event存在少量结构差异，可收集所有出现过的字段：

from pyspark.sql.functions import explode, map_keys, collect_set

# 提取所有event中出现过的字段名
all_event_fields = filtered_df.select(explode(map_keys(col("event")))) \
                              .agg(collect_set("col")) \
                              .first()[0]

步骤2：重构DataFrame列 #

保留顶级非event字段，同时用提取到的字段重构event列：

from pyspark.sql.functions import struct

# 构建event下的列（重命名为原字段名）
event_subcols = [col(f"event.{field}").alias(field) for field in event_field_names]
# 用struct重新组合event列
new_event_col = struct(*event_subcols).alias("event")

# 保留其他顶级列，加上重构后的event列
other_top_cols = [col(col_name) for col_name in filtered_df.columns if col_name != "event"]
trimmed_schema_df = filtered_df.select(*other_top_cols, new_event_col)

注意事项 #

• 方法一操作简单，但如果数据量极大，toJSON会有一定性能开销，适合中小规模过滤后数据集。
• 方法二更高效，精准针对嵌套字段处理，适合仅部分嵌套结构需要精简的场景。
• 若过滤后存在大量event为null的行，记得先过滤掉这些行再提取样本字段，避免获取空结构。

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