将深度嵌套JSON展平为多行并转CSV（12GB大文件内存优化）

需要将12GB的深度嵌套JSON转换为指定列的CSV，样本数据如下：

{
    'reporting_entity_name':'Blue Cross and Blue Shield of Alabama',
    'reporting_entity_type':'health insurance issuer',
    'last_updated_on':'2022-06-10',
    'version':'1.1.0',
    'in_network':[
        {
            'negotiation_arrangement': 'ffs',
            'name': 'xploration of Kidney',
            'billing_code_type': 'CPT',
            'billing_code_type_version': '2022',
            'billing_code': '50010',
            'description': 'Renal Exploration, Not Necessitating Other Specific Procedures',
            'negotiated_rates': [
                {
                    'negotiated_prices': [
                        {'negotiated_type': 'negotiated', 'negotiated_rate': 993.0, 'expiration_date': '2022-06-30', 'service_code': ['21', '22', '24'], 'billing_class': 'professional'},
                        {'negotiated_type': 'negotiated', 'negotiated_rate': 1180.68, 'expiration_date': '2022-06-30', 'service_code': ['21', '22', '24'], 'billing_class': 'professional'},
                        # 其余价格项省略
                    ]
                }
            ]
        }
    ]
}

期望输出CSV包含以下列：

{
    'reporting_entity_name':'',
    'reporting_entity_type':'',
    'last_updated_on':'',
    'version':'',
    'negotiation_arrangement':'',
    'name':'',
    'billing_code_type':'',
    'billing_code_type_version':'',
    'billing_code':'',
    'description':'',
    'provider_groups':'',
    'negotiated_type':'',
    'negotiated_rate':'',
    'expiration_date':'',
    'service_code':'',
    'billing_class':''
}

已尝试pandas的json_normalize、flatten方法及自定义模块，但仅能展平列无法生成对应行；因数据量巨大，担心递归嵌套循环会耗尽内存，寻求高效解决方案。

针对大文件场景，采用流式解析+逐行写入的方式，避免一次性加载整个JSON到内存，核心思路如下：

• 用ijson库流式读取JSON，逐段解析嵌套结构
• 提取顶层公共字段，遍历嵌套数组生成对应行
• 用Python内置csv模块逐行写入CSV，严格控制内存占用

代码实现 #

import ijson
import csv

# 定义输出CSV的列顺序，与期望列完全匹配
output_columns = [
    'reporting_entity_name', 'reporting_entity_type', 'last_updated_on', 'version',
    'negotiation_arrangement', 'name', 'billing_code_type', 'billing_code_type_version',
    'billing_code', 'description', 'provider_groups', 'negotiated_type', 'negotiated_rate',
    'expiration_date', 'service_code', 'billing_class'
]

# 打开JSON源文件和CSV输出文件
with open('large_data.json', 'r') as json_file, open('output.csv', 'w', newline='') as csv_file:
    writer = csv.DictWriter(csv_file, fieldnames=output_columns)
    writer.writeheader()
    
    # 流式解析JSON内容
    parser = ijson.parse(json_file)
    top_fields = {}
    current_in_network = None

    for prefix, event, value in parser:
        # 提取顶层公共字段
        if prefix in ['reporting_entity_name', 'reporting_entity_type', 'last_updated_on', 'version']:
            top_fields[prefix] = value
        # 开始解析in_network下的单个条目
        elif prefix == 'in_network.item' and event == 'start_map':
            current_in_network = {}
        # 收集in_network条目的字段
        elif prefix.startswith('in_network.item.') and event != 'end_map':
            key = prefix.split('.')[-1]
            current_in_network[key] = value
        # 完成单个in_network条目的解析，生成CSV行
        elif prefix == 'in_network.item' and event == 'end_map':
            # 遍历negotiated_rates和negotiated_prices嵌套数组
            for rate_group in current_in_network.get('negotiated_rates', []):
                for price in rate_group.get('negotiated_prices', []):
                    # 展开service_code数组，每个code单独生成一行
                    for service_code in price.get('service_code', []):
                        row = top_fields.copy()
                        row.update({
                            'negotiation_arrangement': current_in_network.get('negotiation_arrangement'),
                            'name': current_in_network.get('name'),
                            'billing_code_type': current_in_network.get('billing_code_type'),
                            'billing_code_type_version': current_in_network.get('billing_code_type_version'),
                            'billing_code': current_in_network.get('billing_code'),
                            'description': current_in_network.get('description'),
                            'provider_groups': '',  # 样本无此字段，留空
                            'negotiated_type': price.get('negotiated_type'),
                            'negotiated_rate': price.get('negotiated_rate'),
                            'expiration_date': price.get('expiration_date'),
                            'service_code': service_code,
                            'billing_class': price.get('billing_class')
                        })
                        writer.writerow(row)
            # 及时清理临时变量，释放内存
            del current_in_network
            current_in_network = None

关键优化点 #

1. 流式解析：ijson仅加载当前解析的JSON片段，内存占用仅取决于单个嵌套节点的大小，不会加载12GB完整文件
2. 逐行写入：每生成一行数据立即写入CSV，无需缓存所有行
3. 数组展开：针对service_code数组，每个code单独生成一行，符合CSV的行式结构要求
4. 内存回收：处理完单个in_network条目后立即删除临时变量，避免内存累积

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