读取扁平化10万+JSON文件耗时过长 求多进程等加速优化方案

首先你贴的代码有个非常影响性能的低级错误：外层套了完全冗余的for i in range(len(json_files))循环，等于整个文件列表会被重复遍历N次（N为文件总数），10万级文件量下这部分会直接把耗时拉到离谱的程度，第一步先把这层多余循环删掉。
单进程串行处理的模式本身也不适合10万级文件的批量任务：整个流程包含磁盘IO读文件、JSON解析、数据扁平化计算三类操作，串行执行时IO等待、CPU计算的开销会累计，整体耗时自然很高。

多进程实现代码 #

直接用Python标准库concurrent.futures.ProcessPoolExecutor实现即可，注意单文件处理逻辑要定义成顶层函数，避免多进程序列化报错，进程数默认取CPU核心数即可，不要盲目开太多进程增加调度开销：

import os
import json
from concurrent.futures import ProcessPoolExecutor, as_completed

# 单文件处理逻辑，必须定义在顶层才能被多进程正常序列化
def process_single_json(file_path, filter_key):
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            jtext = json.load(f)
        # 提前判断目标键是否存在，避免KeyError中断整个批量任务
        if filter_key not in jtext:
            return None
        return flatten_data(jtext[filter_key])
    except Exception as e:
        print(f"处理失败文件: {file_path}, 错误信息: {str(e)}")
        return None

if __name__ == '__main__':
    path_to_json = "替换成你的JSON文件目录"
    target_key = "KEY_TO_FILTER"
    # 主进程提前拼好所有文件完整路径，避免子进程重复做路径计算
    json_files = [
        os.path.join(path_to_json, f) 
        for f in os.listdir(path_to_json) 
        if f.endswith('.json')
    ]
    result = []
    # SSD场景可以设为 os.cpu_count() * 2，机械盘建议保持默认CPU核心数
    worker_num = os.cpu_count()
    with ProcessPoolExecutor(max_workers=worker_num) as executor:
        task_map = {
            executor.submit(process_single_json, file_path, target_key): file_path 
            for file_path in json_files
        }
        for future in as_completed(task_map):
            res = future.result()
            if res is not None:
                result.append(res)

其他提速建议 #

• 替换JSON解析库：把标准库json换成第三方库orjson，解析速度是标准库的3-5倍，能大幅降低JSON解析的CPU开销，替换成本极低，仅需要调整加载逻辑：

  import orjson
  # 替换原来的json.load逻辑
  with open(file_path, 'rb') as f:
      jtext = orjson.loads(f.read())
  

• 逻辑前置减负：不要对全量JSON数据做扁平化，先取出你需要的KEY_TO_FILTER对应的值，再做扁平化处理，直接减少后续计算的数据量
• 合理设置进程数：如果存储是机械硬盘，不要把进程数开太大，机械盘随机IO性能差，过多进程同时读会导致磁头频繁寻道，反而降低读取效率；SSD场景可以把进程数调到CPU核心数的1.5-2倍，充分利用IO带宽
• 跳过无效文件：提前加异常捕获，单个文件损坏、格式错误不要中断整个批量任务，所有文件处理完后再统一排查报错的文件即可
• 如果内存足够，可以考虑批量读取的时候适当调大文件读取缓冲区，减少磁盘IO次数

Windows环境运行多进程代码必须把主执行逻辑放在if __name__ == '__main__':块内，否则会出现进程递归启动的报错。

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