如何将含嵌套列表的生成器输出（字典类型）保存为JSON格式

嘿，这个问题我之前处理大数据集的时候也碰到过，内存不够真的头疼！咱们一步步来解决，保证不用把所有数据都塞进内存里：

首先得明确你的核心需求：把每个生成器批次输出里的fromdata列表元素全部合并成一个大列表，同时保留其他固定参数（比如param0、param1这些）对吧？如果是这样，我们可以用流式写入的思路，分批次把数据写到文件里，完全不用加载全部数据。

方案一：原生Python实现（不需要额外库） #

这个方法最接地气，不用装任何第三方包，新手也能快速上手。思路是先写出JSON的头部（包含那些固定参数），然后逐个批次把fromdata的元素追加到数组里，最后收尾闭合JSON结构。

import json

def save_generator_output(generator, output_file):
    # 先取第一个批次的输出，分离固定参数和数据
    first_batch = next(generator)
    fixed_settings = {k: v for k, v in first_batch.items() if k != "fromdata"}
    first_data = first_batch["fromdata"]

    with open(output_file, "w", encoding="utf-8") as f:
        # 先写固定参数部分，手动拼接fromdata数组的开头
        # 把固定参数序列化后去掉末尾的}，再加上, "fromdata": [
        f.write(json.dumps(fixed_settings)[:-1])
        f.write(', "fromdata": [')
        
        # 写入第一个批次的元素
        if first_data:
            for idx, item in enumerate(first_data):
                if idx > 0:
                    f.write(",")
                json.dump(item, f)
        
        # 遍历剩下的所有批次，逐个写入元素
        for batch in generator:
            for item in batch["fromdata"]:
                f.write(",")
                json.dump(item, f)
        
        # 闭合整个JSON结构
        f.write("]}")

特殊情况处理：如果参数随批次变化 #

如果你的param0、param1这些参数在不同批次里会变化，那上面的方法就不适用了——得换成下面这种每个批次都保存完整字典的写法：

def save_full_batches_to_json(generator, output_file):
    with open(output_file, "w", encoding="utf-8") as f:
        f.write("[")
        is_first_batch = True
        for batch in generator:
            if not is_first_batch:
                f.write(",")
            json.dump(batch, f)
            is_first_batch = False
        f.write("]")

这个写法会把每个批次的完整字典存成一个JSON数组，同样是流式写入，内存占用极低。

方案二：用json-stream库简化操作 #

如果你想用更优雅的方式处理流式JSON，可以试试json-stream这个专门的库，它能帮你省去手动拼接字符串的麻烦。

先安装库：

pip install json-stream

然后写代码：

from json_stream import streamable, dumps
from json_stream.dump import dump_to_file

@streamable
def build_streamed_output(generator):
    # 先拿第一个批次的固定参数
    first_batch = next(generator)
    # 返回固定参数的键值对
    yield ("param0", first_batch["param0"])
    yield ("param1", first_batch["param1"])
    
    # 定义一个流式生成器，输出所有fromdata的元素
    def stream_all_data():
        # 先输出第一个批次的数据
        yield from first_batch["fromdata"]
        # 再输出剩下所有批次的数据
        for batch in generator:
            yield from batch["fromdata"]
    
    # 把流式数据关联到fromdata键
    yield ("fromdata", stream_all_data())

# 调用方法：传入你的装饰器后的生成器，指定输出文件
your_generator = your_decorated_function()  # 替换成你的生成器
dump_to_file(build_streamed_output(your_generator), "output.json")

这个方法的优势是不用手动处理JSON的结构拼接，库会自动帮你处理流式写入，代码看起来更整洁。

内容来源于stack exchange