Python多进程中优化器迭代DataFrame存储方案咨询

针对你在多进程优化器流程中需要安全记录DataFrame的需求，以下是几种实用的解决方案：

1. 集中式队列+单独写入进程 #

让子进程专注于计算，将生成的DataFrame序列化后放入进程安全队列，由单独的写入进程统一处理IO操作，彻底避免多进程直接IO竞争。

实现示例 #

import multiprocessing as mp
import pandas as pd
import pyarrow as pa
import nevergrad as ng
from concurrent import futures

def writer_process(queue):
    # 初始化持久化文件，用parquet格式高效存储
    with open('optimization_results.parquet', 'ab') as f:
        while True:
            data = queue.get()
            if data is None:  # 接收终止信号
                break
            # 反序列化DataFrame
            df = pa.deserialize(data)
            df.to_parquet(f, append=True)

# 修改run_model，返回损失值和需要记录的DataFrame
def run_model(params):
    # 你的原有计算逻辑
    loss = ...  # 优化目标值
    df = ...    # 需要记录的DataFrame
    return loss, df

# 回调函数：将任务结果中的DataFrame序列化后放入队列
def task_callback(result):
    _, df = result
    serialized_df = pa.serialize(df).to_buffer().to_pybytes()
    queue.put(serialized_df)

# 主进程逻辑
if __name__ == "__main__":
    instrum = ...  # 你的参数化配置
    queue = mp.Queue(maxsize=50)  # 设置队列大小，避免内存溢出
    writer = mp.Process(target=writer_process, args=(queue,))
    writer.start()

    optimizer = ng.optimizers.NGOpt(parametrization=instrum, budget=10000, num_workers=25)
    with futures.ProcessPoolExecutor(max_workers=optimizer.num_workers) as executor:
        recommendation = optimizer.minimize(
            run_model, 
            verbosity=0, 
            executor=executor, 
            batch_mode=False,
            callbacks=[task_callback]
        )

    # 所有任务完成后，终止写入进程
    queue.put(None)
    writer.join()

优缺点 #

• 优点：IO操作完全隔离，避免多进程竞争导致的崩溃；子进程专注计算，性能不受IO影响；支持实时监控（写入进程可以同时更新监控用的临时存储）。
• 缺点：需要处理DataFrame的序列化/反序列化；队列满时会阻塞子进程，需根据机器内存调整队列大小。

2. 优化SQLite3使用方式 #

你之前尝试的SQLite3并非不能用，只需开启WAL（Write-Ahead Logging）模式，即可支持多进程并发读写，大幅降低崩溃概率。

实现示例 #

import sqlite3
import pandas as pd
import nevergrad as ng
from concurrent import futures

def run_model(params):
    # 原有计算逻辑
    loss = ...
    df = ...

    # 每个进程创建独立连接并开启WAL模式
    conn = sqlite3.connect('optimization_data.db')
    conn.execute('PRAGMA journal_mode=WAL;')
    # 追加数据到SQLite表
    df.to_sql('results', conn, if_exists='append', index=False)
    conn.close()
    return loss

if __name__ == "__main__":
    instrum = ...
    optimizer = ng.optimizers.NGOpt(parametrization=instrum, budget=10000, num_workers=25)
    with futures.ProcessPoolExecutor(max_workers=optimizer.num_workers) as executor:
        recommendation = optimizer.minimize(
            run_model, 
            verbosity=0, 
            executor=executor, 
            batch_mode=False
        )

优缺点 #

• 优点：无需额外引入依赖，兼容原有逻辑；WAL模式下支持并发读写，稳定性大幅提升。
• 缺点：SQLite3的写入吞吐量有限，25个进程并发写入时可能出现轻微延迟。

3. 按进程拆分临时文件，最后合并 #

让每个子进程写入专属的临时文件，完全避免进程间IO竞争，所有迭代完成后再合并为统一的DataFrame。

实现示例 #

import os
import pandas as pd
from glob import glob
import nevergrad as ng
from concurrent import futures

def run_model(params):
    # 原有计算逻辑
    loss = ...
    df = ...

    # 按进程ID生成专属文件名
    pid = os.getpid()
    df.to_parquet(f'temp_data_{pid}.parquet', if_exists='append', index=False)
    return loss

# 合并所有临时文件的函数
def merge_temp_files():
    temp_files = glob('temp_data_*.parquet')
    combined_df = pd.concat([pd.read_parquet(f) for f in temp_files], ignore_index=True)
    combined_df.to_parquet('final_results.parquet', index=False)
    # 清理临时文件
    for f in temp_files:
        os.remove(f)

if __name__ == "__main__":
    instrum = ...
    optimizer = ng.optimizers.NGOpt(parametrization=instrum, budget=10000, num_workers=25)
    with futures.ProcessPoolExecutor(max_workers=optimizer.num_workers) as executor:
        recommendation = optimizer.minimize(
            run_model, 
            verbosity=0, 
            executor=executor, 
            batch_mode=False
        )
    merge_temp_files()

优缺点 #

• 优点：实现简单，完全无进程竞争，不会出现崩溃；适合大规模迭代场景。
• 缺点：实时监控需要额外读取多个临时文件合并，不如集中式队列方便。

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