调用生成百万级数据的Python函数触发MemoryError,求无修改解决方案
解决百万级数据返回导致MemoryError的方案
不用修改get_data函数,我们可以通过临时替换列表的行为,让函数生成数据的过程中直接写入CSV,避免一次性加载所有数据到内存。以下是两种针对性的实现方法:
方法一:拦截列表的append/extend操作(适用于函数内部用append构建列表的场景)
如果get_data是通过循环调用list.append()或list.extend()来生成结果列表的,我们可以临时替换这两个方法,让每次添加元素时直接写入CSV,不将数据存到列表中:
import csv import contextlib from afs_module import get_data @contextlib.contextmanager def redirect_list_to_csv(csv_path): # 保存列表原有的append和extend方法 original_append = list.append original_extend = list.extend with open(csv_path, 'w', newline='') as csv_file: writer = csv.writer(csv_file) # 重写append:直接写入CSV,不存储到列表 def custom_append(self, item): writer.writerow(item) # 重写extend:批量写入CSV def custom_extend(self, items): writer.writerows(items) # 替换全局列表的方法 list.append = custom_append list.extend = custom_extend try: yield finally: # 恢复原方法,避免影响其他代码 list.append = original_append list.extend = original_extend # 使用示例 with redirect_list_to_csv('output_data.csv'): # 调用get_data时,内部的列表操作都会直接写CSV,返回的是空列表 _ = get_data()
方法二:替换全局list类(适用于函数用列表推导式或直接生成列表的场景)
如果get_data是通过列表推导式(如return [x for x in big_data_source])直接生成列表的,上面的方法会失效,此时可以临时替换全局的list类,让新创建的列表在初始化时就把数据写入CSV,自身不存储数据:
import csv import builtins from afs_module import get_data class CSVWriterList(list): def __init__(self, iterable=None, csv_path='output_data.csv'): # 初始化时直接将迭代器中的数据写入CSV if iterable is not None: with open(csv_path, 'w', newline='') as csv_file: writer = csv.writer(csv_file) writer.writerows(iterable) # 调用父类初始化,但不存储任何元素 super().__init__() # 临时替换全局list类 original_list = builtins.list builtins.list = CSVWriterList try: # get_data返回的是CSVWriterList实例,数据已写入文件,自身为空 _ = get_data() finally: # 恢复原list类 builtins.list = original_list
注意事项
- 如果返回的每个数据是字典类型,需要将
csv.writer替换为csv.DictWriter,并指定fieldnames参数。 - 临时替换全局列表操作可能会影响
get_data内部其他无关的列表使用,建议测试后再投入生产环境。 - 如果
get_data内部有复杂的列表操作逻辑,可能需要根据实际情况调整拦截逻辑。
内容的提问来源于stack exchange,提问作者Rita jhonson
相关产品推荐
相关产品推荐

