Python生成器未降低内存消耗，求实现排查及优化方案

我有三个json.gz文件，分别存储按不同ID分组的餐厅、菜单和匹配数据。需读取所有文件并按ID生成包含对应三类数据的新JSON文件，但运行代码的虚拟机存在内存限制。最初我将三个文件全部加载为JSON对象后用普通for循环处理，导致内存消耗过高。意识到仅需单个ID对应的数据来生成最终文件，不应同时加载所有ID的数据，因此采用Python生成器方案并编写了如下代码：

from json import loads

def load_data_set(string_restaurants_data: str,
                  string_menus_data: str,
                  string_matchings_data: str,): 


  menus_data = loads(string_menus_data)   
  matchings_data = loads(string_matchings_data)   
  restaurants_data = loads(string_restaurants_data)
  
  for id, menu_data in menus_data.items():
    yield id, restaurants_data[id], menu_data, matchings_data[id] if id in matchings_data else "{}"



def main():
    '''
    some code to read the json.gz files. The data is stored in these three string variables: string_restaurants_data, string_menus_data, string_matchings_data
    '''

    restaurants_data_set = load_data_set(string_restaurants_data,
                                             string_menus_data,
                                             string_matchings_data)
    
    size_generator = sys.getsizeof(restaurants_data_set)

    del string_menus_data
    del string_restaurants_data
    del string_matchings_data
    gc.collect()

    list_result = {}
    for restaurant in restaurants_data_set:
      result_data = aggregate_menu_data(restaurant[0],
                                        dumps(restaurant[2]),
                                        dumps(restaurant[1]),
                                        dumps(restaurant[3]),
                                        string_parameters,
                                        eval(debug))
    

      list_result.update(result_data)
    
    data['result'] = dumps(list_result)

但对比无生成器的版本，虚拟机任务管理器显示内存消耗未明显降低。现咨询：该Python生成器实现是否正确？有无更优方案可按需加载所需数据且不影响处理速度？Python版本为3.11。

一、当前生成器实现的问题 #

你的生成器实现不正确，核心问题在于：

• 在load_data_set函数内部，你已经把三个JSON字符串全部加载成了完整的字典对象（menus_data、matchings_data、restaurants_data），这三个大字典已经完全占用了内存，生成器只是逐个返回其中的元素，并没有减少内存占用——生成器本身占用内存极小，但背后的三个完整数据集已经把内存占满了，所以和无生成器版本内存消耗差异不大。
• 你在生成器创建后才删除原始字符串并触发GC，但三个大字典已经存在于内存中，这部分内存并没有被释放。

二、优化方案 #

要真正实现按需加载、降低内存占用，需要从「读取文件」和「解析JSON」两个环节入手，避免一次性加载整个文件内容或解析整个JSON结构：

1. 流式读取并解析JSON（推荐） #

因为你的JSON是按ID分组的键值对结构，可以使用ijson库（需先安装：pip install ijson）来流式解析JSON，逐个读取每个ID对应的条目，而不是一次性加载整个JSON到内存：

import ijson
import gzip
from json import dumps

def stream_id_entries(file_path):
    """流式读取gzip压缩的JSON文件，逐个返回(ID, 对应数据)"""
    with gzip.open(file_path, 'rb') as f:
        # 解析JSON对象的每个键值对
        parser = ijson.kvitems(f, '')
        for id, data in parser:
            yield id, data

def main():
    # 三个文件的流式生成器
    menus_stream = stream_id_entries('menus.json.gz')
    # 匹配数据量通常较小，可预加载到字典按需查找
    matchings_dict = dict(stream_id_entries('matchings.json.gz'))

    list_result = {}
    # 遍历菜单的每个ID，按需获取对应餐厅数据
    for menu_id, menu_data in menus_stream:
        # 重新打开餐厅流，查找当前ID对应的数据（生成器只能遍历一次）
        restaurant_data = {}
        with gzip.open('restaurants.json.gz', 'rb') as f:
            restaurant_parser = ijson.kvitems(f, '')
            for id, data in restaurant_parser:
                if id == menu_id:
                    restaurant_data = data
                    break

        matching_data = matchings_dict.get(menu_id, {})
        # 调用聚合函数（建议直接传Python对象，避免重复序列化）
        result_data = aggregate_menu_data(
            menu_id,
            menu_data,
            restaurant_data,
            matching_data,
            string_parameters,
            bool(eval(debug))  # 提前解析debug参数，避免重复eval
        )
        list_result.update(result_data)

    # 若最终结果过大，可改为逐个写入文件，避免占用过多内存
    data['result'] = dumps(list_result)

2. 进阶优化：同步遍历排序后的流 #

如果餐厅数据量极大，每次重新打开流查找效率太低，可以先提取所有ID并排序，再同步遍历三个流：

import ijson
import gzip
from json import dumps

def get_sorted_ids(file_path):
    """提取文件中所有ID并排序"""
    ids = []
    with gzip.open(file_path, 'rb') as f:
        parser = ijson.keys(f, '')
        for id in parser:
            ids.append(id)
    return sorted(ids)

def stream_sorted_entries(file_path, sorted_ids):
    """按指定ID顺序流式返回对应数据"""
    with gzip.open(file_path, 'rb') as f:
        parser = ijson.kvitems(f, '')
        id_iter = iter(sorted_ids)
        current_target_id = next(id_iter, None)
        for id, data in parser:
            while current_target_id is not None and id > current_target_id:
                # 若当前ID大于目标ID，说明目标ID不存在，返回空数据
                yield current_target_id, {}
                current_target_id = next(id_iter, None)
            if id == current_target_id:
                yield id, data
                current_target_id = next(id_iter, None)
        # 处理剩余未找到的ID
        while current_target_id is not None:
            yield current_target_id, {}
            current_target_id = next(id_iter, None)

def main():
    # 获取菜单的所有ID并排序（假设菜单ID是全集）
    sorted_ids = get_sorted_ids('menus.json.gz')

    # 按排序后的ID同步遍历三个流
    restaurants_stream = stream_sorted_entries('restaurants.json.gz', sorted_ids)
    menus_stream = stream_sorted_entries('menus.json.gz', sorted_ids)
    matchings_stream = stream_sorted_entries('matchings.json.gz', sorted_ids)

    list_result = {}
    for (rid, rdata), (mid, mdata), (maid, madata) in zip(restaurants_stream, menus_stream, matchings_stream):
        # 三个流的ID应该一致
        result_data = aggregate_menu_data(
            rid,
            mdata,
            rdata,
            madata,
            string_parameters,
            bool(eval(debug))
        )
        list_result.update(result_data)

    data['result'] = dumps(list_result)

3. 额外内存优化细节 #

• 避免全量存储结果：如果最终生成的JSON文件过大，不要把所有结果存在list_result字典中，而是每次生成一条结果就追加写入输出文件（注意处理JSON格式的首尾括号）。
• 减少序列化开销：你的代码中把数据dump成字符串传给aggregate_menu_data，如果函数内部需要反序列化，会浪费性能，建议直接传递Python对象。
• 替换eval(debug)：eval存在安全风险，且重复调用会降低性能，建议提前解析为布尔值，比如debug_flag = bool(eval(debug))，再传递给函数。

三、内存占用验证方法 #

可以使用tracemalloc模块精准跟踪内存使用，比任务管理器更可靠：

import tracemalloc

tracemalloc.start()

# 执行你的业务代码

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')

print("[Top 10内存占用条目]")
for stat in top_stats[:10]:
    print(stat)

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