多进程处理SQLAlchemy模型时内存不足（code137）问题求助

一、解决内存溢出（退出码137）与BrokenProcessPool异常 #

1. 避免全量加载ORM对象到内存 #

350万ORM实例直接存放在列表里本身就会占用巨量内存，多进程启动时，子进程会复制父进程的内存空间（写时复制机制下，只要有修改就会触发全量复制），直接导致内存爆炸。

• 改用分批查询+处理：从原数据库分批拉取数据，每次处理1000-5000条，避免一次性加载全量数据。示例代码：

from sqlalchemy.orm import sessionmaker

OriginalSession = sessionmaker(bind=original_db_engine)
batch_size = 2000

with OriginalSession() as session:
    # 用yield_per和in_batches实现分批查询
    query = session.query(OriginalModel).yield_per(batch_size)
    for batch in query.in_batches(batch_size=batch_size):
        process_batch(batch)

2. 优化多进程的内存与序列化问题 #

ProcessPoolExecutor的默认配置容易引发内存和进程崩溃问题，调整方式如下：

• 限制进程数量：根据容器内存调整max_workers，比如4核机器设置为2，减少同时运行的进程数。
• 传递原始数据而非ORM对象：ORM实例绑定了父进程的会话，跨进程序列化不仅占内存，还会导致进程池异常。只传递需要的字段数据（比如字典或元组），在子进程内转换为FooModel：

from concurrent.futures import ProcessPoolExecutor

def process_item(raw_data):
    # 子进程内处理数据并生成FooModel
    foo = FooModel()
    foo.field1 = raw_data["field1"]
    # 其他字段预处理逻辑
    return foo

# 父进程中提取需要的字段，避免传递ORM内部属性
batch_raw = [{k: getattr(obj, k) for k in ["field1", "field2"]} for obj in batch]

with ProcessPoolExecutor(max_workers=2) as executor:
    processed_foos = list(executor.map(process_item, batch_raw))

3. 修复BrokenProcessPool异常 #

这个异常本质是子进程崩溃（多数是OOM被系统杀死）导致的，解决内存问题后通常会消失。额外注意：

• 子进程独立创建数据库会话：不要共享父进程的会话，每个子进程自己初始化目标数据库的会话。
• 子进程内添加异常捕获：避免单个任务崩溃导致整个进程池挂掉。

二、并行化批量写入目标数据库 #

针对load_in_another_database()的并行需求，核心是批量操作+进程隔离：

• 用批量插入代替单条写入：每处理完1000条左右的FooModel，用bulk_save_objects或bulk_insert_mappings批量写入，减少数据库IO开销。
• 子进程独立管理写入会话：每个子进程单独创建目标数据库的会话，避免跨进程共享连接引发的问题。示例代码：

def process_and_save(raw_batch):
    TargetSession = sessionmaker(bind=target_db_engine)
    with TargetSession() as target_session:
        foos = []
        for raw_data in raw_batch:
            foo = FooModel()
            foo.field1 = raw_data["field1"]
            # 预处理逻辑
            foos.append(foo)
        # 批量插入
        target_session.bulk_save_objects(foos)
        target_session.commit()

# 父进程拆分批次，传给子进程处理
with ProcessPoolExecutor(max_workers=2) as executor:
    split_batches = [batch_raw[i:i+1000] for i in range(0, len(batch_raw), 1000)]
    executor.map(process_and_save, split_batches)

三、额外内存优化技巧 #

• 只加载需要的字段：查询原数据时，用query.with_entities(OriginalModel.field1, OriginalModel.field2)只拉取需要的字段，避免加载ORM对象的关联属性和冗余数据。
• 手动触发垃圾回收：每处理完一批数据后，调用gc.collect()强制回收内存，尤其是在分批循环中。
• 使用轻量数据结构：用元组传递字段数据，比字典更节省内存，比如(obj.field1, obj.field2)。

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