一次性序列化压缩大对象至文件内存不足，如何分块处理？

核心思路 #

不用一次性把所有数据加载到内存，通过分块序列化+流式压缩写入的方式逐块处理，避免同时持有多份完整数据，从根源上解决内存溢出问题。

具体实现方案 #

1. 分批次序列化对象列表 #

不直接序列化整个objects列表，而是按小批次遍历处理，内存中仅保留当前批次的对象和序列化后的数据块，而非整个超大列表。

2. 增量压缩+即时写入 #

使用brotli提供的BrotliCompressor类，它支持分多次喂入数据，每次压缩后立即将结果写入文件，无需等待所有数据压缩完成再统一保存。

完整代码示例 #

import brotli
import dill

def save_large_objects(objects, file_path, batch_size=100, quality=1):
    with open(file_path, 'wb') as output, brotli.BrotliCompressor(quality=quality) as compressor:
        # 按批次遍历处理对象列表
        for i in range(0, len(objects), batch_size):
            batch = objects[i:i+batch_size]
            # 序列化当前批次的对象
            batch_serialized = dill.dumps(batch, dill.HIGHEST_PROTOCOL)
            # 增量压缩并写入文件
            compressed_chunk = compressor.process(batch_serialized)
            if compressed_chunk:
                output.write(compressed_chunk)
        # 处理压缩收尾阶段的剩余数据
        final_chunk = compressor.finish()
        if final_chunk:
            output.write(final_chunk)

# 使用示例
# save_large_objects(objects, "large_objects_compressed.br")

内存优化原理 #

• 序列化阶段：每次仅处理batch_size个对象，内存中仅留存当前批次的对象和对应的序列化小数据块，不会加载整个超大列表。
• 压缩写入阶段：process方法会实时返回已压缩的数据块，写入后即可释放这部分内存，无需缓存全部压缩结果。

额外优化建议 #

• 调整batch_size：根据单个对象的大小和可用内存灵活设置，对象越大，批次尺寸应越小。
• 及时释放无用内存：处理完原始objects列表后，用del objects主动释放内存。
• 数据完整性验证：写入完成后可通过流式解压验证数据是否正常：

  def load_large_objects(file_path):
      objects = []
      with open(file_path, 'rb') as input_file, brotli.BrotliDecompressor() as decompressor:
          while True:
              # 按固定大小读取压缩文件
              chunk = input_file.read(4096)
              if not chunk:
                  break
              decompressed_data = decompressor.process(chunk)
              if decompressed_data:
                  # 反序列化批次并合并到总列表
                  batch = dill.loads(decompressed_data)
                  objects.extend(batch)
          # 处理解压收尾的剩余数据
          final_decompressed = decompressor.finish()
          if final_decompressed:
              batch = dill.loads(final_decompressed)
              objects.extend(batch)
      return objects
  

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