ProcessPoolExecutor与ThreadPoolExecutor性能差异及代码优化问询

1. 进程间数据拷贝的巨大开销
   你的任务中需要传递X_val（10万行×1000列）、训练好的模型fit等大对象给子进程。由于进程拥有独立内存空间，这些对象需要通过pickle序列化后完整拷贝到子进程，这个过程的耗时远超过并行计算带来的收益。而线程池共享父进程内存，无需额外拷贝，因此整体开销更低。
2. sklearn底层的GIL释放机制
   虽然Python的GIL会限制单线程CPU利用率，但RandomForestClassifier.predict()底层依赖的是C实现的计算逻辑（如OpenMP），这些代码会主动释放GIL，允许线程池中的多个线程同时利用多核CPU。因此线程池在你的CPU密集型任务中也能有效并行，而进程池的额外开销反而拖慢了速度。

1. 参数不匹配
   randomShuffle函数定义的参数是colname, X, y, fit，但在executor.submit()中额外传递了labels=classifier.classes_，这会触发TypeError（函数不接受该参数），需要删除多余的labels参数。
2. 不必要的深拷贝
   X_ = X.copy(deep=True)会深拷贝整个DataFrame，对于10万行的数据集来说，这是极高的性能开销。实际上你只需要修改单个列，完全不需要深拷贝整个数据集。
3. 低效的结果赋值
   在循环中使用acc_scr1.loc[i, ...]逐个赋值，对于大DataFrame来说，这种逐元素操作会频繁触发内存重排，效率极低。
4. 模型传递的冗余开销
   训练好的fit对象在进程池中传递时需要完整序列化，而该对象本身可能包含大量树结构数据，进一步加剧了进程间的拷贝开销。

1. 数据格式与拷贝优化 #

将DataFrame转换为numpy数组（sklearn原生支持numpy输入，且操作更快），同时避免全量拷贝：

def randomShuffle(colname, X_np, y_np, fit, col_idx):
    # 仅复制需要打乱的列所在的数组，其余列直接引用原内存
    X_shuffled = X_np.copy()
    np.random.shuffle(X_shuffled[:, col_idx])
    pred = fit.predict(X_shuffled)
    return {'col_name': colname, 'scr': accuracy_score(y_np, pred)}

在runConcurrent中提前转换并建立列索引映射：

X_val_np = X_val.to_numpy()
y_val_np = y_val.to_numpy()
col_idx_map = {col: idx for idx, col in enumerate(X_val.columns)}

2. 优化线程池使用 #

改用executor.map()替代submit()，减少任务提交的开销，同时批量处理：

# 准备统一的任务参数列表
task_args = [(col, X_val_np, y_val_np, fit, col_idx_map[col]) for col in X_val.columns]
with concurrent.futures.ThreadPoolExecutor() as executor:
    # map批量执行任务，按顺序返回结果
    results = list(executor.map(lambda args: randomShuffle(*args), task_args))
# 一次性将结果赋值到DataFrame，避免逐元素操作的开销
for res in results:
    acc_scr1.loc[i, res['col_name']] = res['scr']

3. 可选：进程池的共享内存优化（如果一定要用进程池） #

使用multiprocessing.shared_memory共享X_val_np，彻底避免进程间的数据拷贝：

from multiprocessing import shared_memory
import numpy as np

# 父进程中创建共享内存，将数据写入共享区域
shm = shared_memory.SharedMemory(create=True, size=X_val_np.nbytes)
X_shared = np.ndarray(X_val_np.shape, dtype=X_val_np.dtype, buffer=shm.buf)
X_shared[:] = X_val_np[:]

# 修改任务函数，从共享内存读取数据
def randomShuffle_shared(colname, shm_name, shape, dtype, y_np, fit, col_idx):
    shm = shared_memory.SharedMemory(name=shm_name)
    X_np = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    X_shuffled = X_np.copy()
    np.random.shuffle(X_shuffled[:, col_idx])
    pred = fit.predict(X_shuffled)
    shm.close()
    return {'col_name': colname, 'scr': accuracy_score(y_np, pred)}

# 提交任务时传递共享内存的元信息
task_args = [(col, shm.name, X_val_np.shape, X_val_np.dtype, y_val_np, fit, col_idx_map[col]) for col in X_val.columns]
with concurrent.futures.ProcessPoolExecutor() as executor:
    results = list(executor.map(lambda args: randomShuffle_shared(*args), task_args))

# 任务结束后释放共享内存
shm.unlink()

4. 其他细节优化 #

• 提前将y_val转换为numpy数组，避免重复的类型转换开销；
• 预先初始化足够大小的acc_scr1，避免循环中动态扩展DataFrame；
• 如果使用线程池，建议将RandomForest模型的n_jobs设为1，避免模型内部并行与外部线程池的CPU资源竞争。

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