如何高效分割超大行式 JSON 文件（5GB+）为多个标准格式 JSON 文件-Python教程-PHP中文网

如何高效分割超大行式 JSON 文件（5GB+）为多个标准格式 JSON 文件

本文介绍一种流式处理方法，绕过全量加载与 json.loads() 解析，直接按行拼接生成符合 { "root": [...] } 结构的分片 json 文件，显著提升 5gb 级行式 json（json lines）的分割性能。

本文介绍一种流式处理方法，绕过全量加载与 json.loads() 解析，直接按行拼接生成符合 { "root": [...] } 结构的分片 json 文件，显著提升 5gb 级行式 json（json lines）的分割性能。

在处理大型行式 JSON（JSON Lines）文件（如 5GB）时，传统方法——先用 f.readlines() 读取全部行、再逐行 json.loads() 解析、最后批量 json.dump() 写入——会导致极高内存占用与解析开销，实测耗时长达 18 分钟。根本瓶颈在于：对每行重复执行完整 JSON 解析/序列化，且将全部对象驻留内存。

更优解是采用 流式字符串拼接 + 零解析（zero-parse）策略：

dmap-jsapi-skill

DMap GL地图JSAPI（dmapgl）开发指南。用于编写、审查、调试 DMap 地图 API 代码，涵盖地图初始化、覆盖物渲染、图层管理、事件处理、控件交互等任务。当用户提及 dmap、dmapgl、jsapi‑gl 或相关地图开发需求时自动触发。

下载

不调用 json.loads() 或 json.load()，避免反序列化开销；
不构建 Python 对象列表，彻底规避内存膨胀；
直接按行读取原始字符串，精准控制缩进与逗号分隔，手动生成合法 JSON；
每达到预设行数（如 50,000 行），立即关闭当前文件并开启新文件。

以下为生产就绪的优化实现：

import os
import glob
import shutil
import time

def split_json_lines(input_path: str, output_dir: str, archive_dir: str, 
                     chunk_size: int = 50000, root_key: str = "ROOT"):
    """
    流式分割 JSON Lines 文件，生成标准 { "ROOT": [...] } 格式分片

    :param input_path: 输入文件路径（每行一个 JSON 对象）
    :param output_dir: 输出目录（分片文件写入位置）
    :param archive_dir: 归档目录（原文件移动目标）
    :param chunk_size: 每个分片包含的 JSON 行数
    :param root_key: 外层根键名，默认 "ROOT"
    """
    os.makedirs(output_dir, exist_ok=True)
    os.makedirs(archive_dir, exist_ok=True)

    # 构建 JSON 模板（避免重复字符串拼接）
    opener = f'{{\n  "{root_key}": [\n'
    closer = '  ]\n}'
    newline = '\n'

    file_index = 0
    row_count = 0
    output_file = None

    try:
        with open(input_path, 'r', encoding='utf-8') as f_in:
            for line_num, line in enumerate(f_in, 1):
                line = line.strip()
                if not line:  # 跳过空行
                    continue

                # 每 chunk_size 行新建一个文件
                if row_count % chunk_size == 0:
                    # 关闭上一个文件（若存在）
                    if output_file:
                        output_file.write(closer)
                        output_file.close()

                    # 创建新文件
                    base_name = os.path.splitext(os.path.basename(input_path))[0]
                    output_path = os.path.join(output_dir, f"{base_name}_{file_index + 1}.json")
                    output_file = open(output_path, 'w', encoding='utf-8')
                    output_file.write(opener)
                    file_index += 1

                # 写入当前 JSON 行：首行为无逗号缩进，后续行前置逗号
                if row_count % chunk_size == 0:
                    output_file.write(f'     {line}')
                else:
                    output_file.write(f'{newline}    ,{line}')

                row_count += 1

        # 写入最后一个文件的闭合括号
        if output_file:
            output_file.write(newline + closer)
            output_file.close()

        # 归档原文件
        shutil.move(input_path, os.path.join(archive_dir, os.path.basename(input_path)))

        print(f"✅ 完成分割：共 {row_count} 行 → {file_index} 个分片（每片 ≤ {chunk_size} 行）")
        return file_index

    except Exception as e:
        if output_file:
            output_file.close()
        raise RuntimeError(f"分割失败：{e}") from e

# 批量处理示例
if __name__ == "__main__":
    start_time = time.time()

    input_pattern = r"C:\Users\JSON\Input\*.json"
    for input_file in glob.glob(input_pattern):
        print(f"? 正在处理：{input_file}")
        split_json_lines(
            input_path=input_file,
            output_dir=input_file.replace(r"C:\Users\JSON\Input", r"C:\Users\JSON\Output"),
            archive_dir=input_file.replace(r"C:\Users\JSON\Input", r"C:\Users\JSON\OriginalFiles_BKP"),
            chunk_size=50000
        )

    end_time = time.time()
    print(f"⏱ 总耗时：{(end_time - start_time):.2f} 秒")