基于Python使用CSV数据更新大体积JSON文件

问题场景 #

有体积超1GB的JSON Lines格式文件（每行一个JSON对象），以及包含ID匹配列的CSV文件，需要通过ID关联给JSON文件中的对应条目新增wikiQRank字段。原方案将JSON转CSV后覆盖的方式效率低下，需用Python实现更优的处理逻辑。

原实现的核心问题 #

• 全量加载大JSON文件到内存，会占用极高内存资源，甚至触发内存溢出
• 用列表存储CSV的ID和qrank数据，通过index()查找匹配项是线性遍历，数据量大时效率极低

优化后的实现方案 #

核心思路 #

1. 将CSV数据转换为ID-qrank字典映射，把查找操作的时间复杂度从O(n)降到O(1)
2. 对大JSON文件采用逐行读写的流式处理，避免全量加载占用内存

优化代码 #

import csv
import json

def load_qrank_map(csv_path):
    # 加载CSV为ID到qrank的字典映射
    qrank_map = {}
    with open(csv_path, 'r', encoding='utf-8') as csv_file:
        reader = csv.reader(csv_file)
        # 跳过CSV表头（如果有表头的话，无表头可删除此行）
        next(reader)
        for row in reader:
            if len(row) >= 2:  # 确保行数据有效
                qrank_map[row[0]] = row[1]
    return qrank_map

def update_json_large_file(input_json_path, output_json_path, qrank_map):
    # 流式处理大JSON文件：逐行读、处理、写
    with open(input_json_path, 'r', encoding='utf-8') as infile, \
         open(output_json_path, 'w', encoding='utf-8') as outfile:
        for line in infile:
            try:
                # 解析单行JSON
                data = json.loads(line.strip())
                # 匹配ID并更新字段
                item_id = data.get('id')
                if item_id in qrank_map:
                    data['wikiQRank'] = qrank_map[item_id]
                # 将处理后的JSON写入新文件
                json.dump(data, outfile, ensure_ascii=False)
                outfile.write('\n')  # 保持JSON Lines格式
            except json.JSONDecodeError:
                # 跳过无效的JSON行
                continue

if __name__ == '__main__':
    # 加载CSV映射
    qrank_map = load_qrank_map('qrank.csv')
    # 处理大JSON文件，输出到新文件（避免直接修改原文件）
    update_json_large_file('enhanced-wikipois', 'updated-enhanced-wikipois', qrank_map)

关键优化点说明 #

• 字典映射：用字典存储CSV的ID与qrank对应关系，查找匹配项的速度大幅提升，尤其适合CSV数据量大的场景
• 流式处理：每次仅处理一行JSON数据，内存占用始终保持在极低水平，完美适配1GB+的大文件
• 容错处理：添加了JSON解析错误的捕获逻辑，跳过无效行避免程序崩溃
• 安全输出：将处理结果写入新文件，避免直接修改原文件导致数据丢失

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