如何用Python流式处理Web服务返回的大JSON并转换为CSV？

哥们，我太懂你这种头疼了——超大JSON一次性加载直接爆内存，json.loads()或者response.json()直接给你扔MemoryError对吧？问题出在你把整个响应都塞进内存里了，对付这种大体积数据，咱们得换流式处理的思路，一点一点啃，别一口吃撑。

核心思路 #

原来的做法是把整个JSON加载成Python对象，再转成DataFrame，这对小数据没问题，但大数据直接把内存拉爆。咱们改成：

1. 流式获取HTTP响应，不一次性下载完整个文件
2. 用ijson库逐个解析JSON数组里的每个对象，而不是全加载到内存
3. 解析一个对象，就把它展开成扁平结构，然后写入CSV一行，全程内存只留当前处理的这一个对象

具体实现步骤 #

1. 先安装依赖 #

需要用到ijson这个流式JSON解析库，还有requests和csv（后者是Python标准库）：

pip install ijson requests

2. 完整代码示例 #

import ijson
import requests
import csv

def flatten_nested_dict(d, parent_key='', sep='_'):
    """递归展开嵌套字典，把F5.F5_1转成F5_F5_1，和pandas.json_normalize效果一致"""
    items = []
    for k, v in d.items():
        new_key = f"{parent_key}{sep}{k}" if parent_key else k
        if isinstance(v, dict):
            items.extend(flatten_nested_dict(v, new_key, sep=sep).items())
        else:
            items.append((new_key, v))
    return dict(items)

def stream_json_to_csv(url, output_file):
    # 流式获取响应，避免一次性下载整个大文件
    with requests.get(url, stream=True) as response:
        response.raise_for_status()  # 检查请求是否成功，失败直接抛异常
        # 从响应原始流中逐个解析JSON数组的元素，'item'表示取数组的每个子对象
        parser = ijson.items(response.raw, 'item')
        
        # 先拿第一个元素确定CSV表头
        first_item = next(parser)
        flattened_first = flatten_nested_dict(first_item)
        fieldnames = flattened_first.keys()
        
        # 打开CSV文件，逐行写入
        with open(output_file, 'w', newline='', encoding='utf-8') as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            # 写入第一个元素
            writer.writerow(flattened_first)
            # 遍历剩下的所有元素，逐个处理写入
            for item in parser:
                flattened_item = flatten_nested_dict(item)
                writer.writerow(flattened_item)

# 替换成你的目标URL和输出文件名
stream_json_to_csv("你的目标API地址", "output.csv")

关键细节解释 #

• stream=True：告诉requests不要一次性把整个响应下载到内存，而是分块获取，内存占用瞬间降下来
• ijson.items(response.raw, 'item')：直接从响应的字节流中解析JSON，全程只在内存保留当前处理的单个JSON对象，完全不会加载整个大数组
• flatten_nested_dict：自己实现的嵌套结构展开函数，完美替代pandas.json_normalize的功能，把嵌套的键（比如F5里的F5_1）转成扁平的CSV列名（F5_F5_1）
• 用csv.DictWriter逐行写入：避免把所有数据放到DataFrame里占用内存，毕竟DataFrame本质还是把所有数据存在内存里的

为啥不用pandas了？ #

不是说pandas不好，而是pandas.json_normalize必须把整个数据集加载到内存才能处理，这正是你遇到MemoryError的根源。换成流式处理后，内存占用几乎可以忽略，只和单个JSON对象的大小有关。

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