Databricks中Python拆分大JSON文件报错及结果异常求助

问题根源 #

• 用Python原生json库读取大文件会将整个文件加载到单节点内存，无法利用集群分布式能力，扩容内存也解决不了本质问题
• 按索引取字段的方式不符合Spark的Schema驱动模型，容易导致生成的DataFrame无有效Schema，触发空Schema错误
• 未针对大字段（如in_network）做分布式拆分处理，写入时易出现内存溢出

修正后的代码方案 #

1. 用Spark分布式读取JSON #

放弃Python原生IO，直接用Spark的read.json处理大文件，自动适配集群分布式能力：

# 读取大JSON文件，自动推断Schema（已知Schema时手动指定更高效）
df = spark.read.json("/path/to/your/large_file.json")

2. 拆分目标DataFrame #

直接通过字段名筛选（JSON是键值结构，索引无实际意义）：

# 第一个文件：包含reporting_entity_name到version的字段
meta_fields = ["reporting_entity_name", "reporting_entity_type", "plan_id", "version"]
meta_df = df.select(*meta_fields)

# 第二个文件：包含in_network及其他非元数据字段
all_fields = df.columns
in_network_fields = [col for col in all_fields if col not in meta_fields]
in_network_df = df.select(*in_network_fields)

3. 优化文件写入 #

• 元数据文件数据量小，用coalesce(1)合并为单个文件
• in_network字段数据量大，用repartition拆分多个文件避免内存溢出：

# 写入元数据文件
meta_df.coalesce(1).write.mode("overwrite").json("/path/to/output/metadata")

# 写入in_network数据，根据实际数据量调整分区数
in_network_df.repartition(10).write.mode("overwrite").json("/path/to/output/in_network")

4. 可选：手动指定Schema提升性能 #

已知JSON结构时，提前定义Schema可避免Spark推断Schema的开销，同时彻底杜绝空Schema问题：

from pyspark.sql.types import StructType, StructField, StringType, ArrayType, MapType

# 示例Schema，根据实际数据结构调整
schema = StructType([
    StructField("reporting_entity_name", StringType(), True),
    StructField("reporting_entity_type", StringType(), True),
    StructField("plan_id", StringType(), True),
    StructField("version", StringType(), True),
    StructField("in_network", ArrayType(MapType(StringType(), StringType())), True),
    # 其他字段按需添加
])

df = spark.read.schema(schema).json("/path/to/your/large_file.json")

原代码错误原因解析 #

• 用Python原生open+json.load读取大文件，属于单节点内存操作，无法利用集群分布式资源，扩容内存只是治标不治本
• 按索引取字段的操作不符合Spark DataFrame的Schema驱动逻辑，易生成无有效Schema的空DataFrame，触发“empty or nested empty schemas”错误
• 未对大字段做分区处理，写入时单节点负载过高，导致内存溢出

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