如何高效将十亿行格式化JSON文件导入Pandas DataFrame？

问题背景 #

你的JSON文件是数组包裹格式（外层为[]，内部是多个JSON对象用,分隔），10亿行规模远超16GB内存的承载能力，直接加载必然内存溢出。下面针对你的场景提供三种最优方案，解决现有方法的痛点。

方案一：Pandas流式分块处理（替代手动分割文件） #

手动分割文件效率低，我们可以通过Python文件流逐块读取JSON数组内容，拆分单个JSON对象后分批导入Pandas，避免一次性加载全部数据。

实现代码 #

import pandas as pd
import json

def json_array_reader(file_path, chunk_size=10000):
    """生成器：逐块读取JSON数组中的对象"""
    with open(file_path, 'r', encoding='utf-8') as f:
        # 跳过开头的[
        next(f)
        buffer = []
        for line in f:
            line = line.strip()
            # 跳过空行和结尾的]
            if not line or line == ']':
                continue
            # 处理结尾的,
            if line.endswith(','):
                line = line[:-1]
            buffer.append(line)
            # 达到chunk_size时返回一批
            if len(buffer) >= chunk_size:
                # 拼接成JSON数组字符串，解析后转DataFrame
                json_str = '[' + ','.join(buffer) + ']'
                yield pd.read_json(json_str)
                buffer = []
        # 处理剩余的对象
        if buffer:
            json_str = '[' + ','.join(buffer) + ']'
            yield pd.read_json(json_str)

# 分批读取并合并成最终DataFrame（如果内存允许合并，否则可逐批处理）
df_list = []
for chunk in json_array_reader('your_file.json', chunk_size=100000):
    df_list.append(chunk)
    # 可选：每批处理后做存储或计算，减少内存占用
    # chunk.to_csv('chunk_xxx.csv', index=False)

final_df = pd.concat(df_list, ignore_index=True)

关键优化点 #

• 用生成器流式读取，每次仅加载chunk_size个JSON对象到内存
• 自动处理数组的[]分隔符，无需手动分割文件
• 可根据内存调整chunk_size（16GB内存建议设为10-50万）

方案二：修复PySpark导入错误，预处理后转Pandas #

PySpark的_corrupt_record错误是因为它默认解析每行一个JSON对象，而你的文件是数组格式。通过文本读取后拆分数组，即可正常解析。

实现代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split, regexp_replace, from_json, col
from pyspark.sql.types import StructType, StructField, StringType

# 初始化SparkSession
spark = SparkSession.builder.appName("BigJSONImport").getOrCreate()

# 1. 读取整个JSON文件为文本（因为是数组格式，不能直接用spark.read.json）
text_df = spark.read.text('your_file.json', wholetext=True)

# 2. 去掉外层的[]，拆分内部的JSON对象
cleaned_df = text_df.select(
    regexp_replace(col('value'), r'^\[|\]$', '').alias('cleaned_value')
).select(
    split(col('cleaned_value'), '\},\s*\{').alias('json_objects')
).select(
    explode(col('json_objects')).alias('json_str')
)

# 3. 修复每个JSON对象的格式（补全首尾的{}）
fixed_df = cleaned_df.select(
    regexp_replace(col('json_str'), r'^', '{').alias('fixed_str')
).select(
    regexp_replace(col('fixed_str'), r'$', '}').alias('final_json')
)

# 4. 定义JSON schema，解析成结构化数据
schema = StructType([
    StructField("TYPE", StringType()),
    StructField("UID", StructType([
        StructField("Number", StringType()),
        StructField("Date", StringType())
    ])),
    StructField("UIDC", StringType())
])

parsed_df = fixed_df.select(
    from_json(col('final_json'), schema).alias('data')
).select('data.*')

# 5. 转成Pandas DataFrame（仅当最终数据量适合内存时执行，否则直接用Spark处理）
pandas_df = parsed_df.toPandas()

# 停止SparkSession
spark.stop()

适用场景 #

适合先对数据做过滤、聚合等预处理，再导出到Pandas的场景，利用Spark的分布式计算能力处理超大规模数据，避免内存瓶颈。

方案三：使用Dask自动分块处理 #

Dask是专为大数据设计的并行计算库，API与Pandas高度兼容，可自动分块读取超出内存的JSON文件。

实现代码 #

import dask.dataframe as dd

# 读取JSON数组，设置blocksize控制分块大小（16GB内存建议设为1GB）
dask_df = dd.read_json('your_file.json', blocksize='1GB')

# 执行计算（如转成Pandas，或直接在Dask上做分析）
pandas_df = dask_df.compute()

优势 #

• 无需手动处理分块逻辑，Dask自动拆分文件
• 支持Pandas大部分API，学习成本低
• 可利用多核CPU加速处理

常见问题解析 #

1. Pandas read_json内存耗尽：一次性加载10亿行数据到内存，远超16GB容量，必须分块处理
2. lines=True报错：lines=True要求每行是独立JSON对象，而你的文件是数组格式，不适用
3. json.load内存耗尽：同Pandas，一次性解析整个JSON数组到内存，必然溢出
4. PySpark _corrupt_record：Spark默认解析单行JSON，数组格式需先转成单个JSON对象再解析

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