使用ProcessPoolExecutor多进程训练多模型遇错求助

问题背景 #

尝试在同一数据集上并行训练多个Sklearn模型（BernoulliNB、RandomForestClassifier等），使用concurrent.futures.ProcessPoolExecutor()时出现进程启动错误，但改用ThreadPoolExecutor()可正常运行却无性能提升。

原代码 #

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score , f1_score , recall_score, precision_score
import time
import concurrent.futures 

df = pd.read_csv("\\.\\DATA\\heart.csv")

X = df.iloc[:,:-1] # We get all but the output column
y = pd.DataFrame(df["output"]) # We get output column

from sklearn.model_selection import train_test_split
x_train, x_test, y_train, y_test = train_test_split(X,y, test_size = 0.15 , random_state = 53)

from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
x_train = scaler.fit_transform(x_train)
x_test = scaler.transform(x_test)


if __name__ == '__main__':

    from sklearn.naive_bayes import BernoulliNB
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.svm import SVC
    from sklearn.linear_model import SGDClassifier

    start = time.perf_counter()

    # list of algorithms names
    algo_name = [BernoulliNB, RandomForestClassifier, SVC, SGDClassifier]

    # function to train the model and print the accuracy 
    def train(algo_name) : 
        model = algo_name().fit(x_train, y_train)
        y_pred = model.predict(x_test)
        acc = accuracy_score(y_test, y_pred)
        f1 = f1_score(y_test, y_pred)
        prec = precision_score(y_test, y_pred)
        recall = recall_score(y_test, y_pred)
        print("--------------------------------")
        print("Accuracy: ", round(accuracy_score(y_test, y_pred)*100, 2), "%")
        print("F1 accuracy: ", round(f1_score(y_test, y_pred)*100, 2), "%")
        print("Precision: ", round(precision_score(y_test, y_pred)*100, 2), "%")
        print("Recall : ", round(recall_score(y_test, y_pred)*100, 2), "%")
        print("--------------------------------")

    # run the train function with every item in the algo_name list parallel to each other
    with concurrent.futures.ProcessPoolExecutor() as executor: 
        executor.map(train, algo_name)

    end = time.perf_counter()
    print(f'Program runtime is {round((end - start) * 1000 , 2)} ms')

报错信息 #

Process SpawnProcess-2:
Process SpawnProcess-4:
Process SpawnProcess-1:
Process SpawnProcess-3:

解决方案 #

问题根源 #

Windows系统下多进程采用spawn启动模式，会重新导入整个脚本，导致全局的数据集预处理代码被重复执行，且进程间无法直接共享NumPy数组这类大型对象；另外y_train用DataFrame格式传入Sklearn模型，可能引发拟合时的维度兼容问题。

修正步骤 #

1. 将数据加载和预处理逻辑移入if __name__ == '__main__':：避免子进程重复执行数据加载，确保只有主进程处理数据。
2. 将标签转为Series格式：Sklearn模型对一维标签的支持更稳定，替换y = pd.DataFrame(df["output"])为y = df["output"]。
3. 让train函数显式接收参数：不再依赖全局变量，将x_train、x_test、y_train、y_test作为参数传入，避免进程间共享变量的问题。
4. （可选）用submit替代map：更灵活地处理任务，方便捕获子进程的异常信息。

修正后的代码 #

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score, recall_score, precision_score
import time
import concurrent.futures 

def train(algo_cls, x_train, x_test, y_train, y_test): 
    model = algo_cls().fit(x_train, y_train)
    y_pred = model.predict(x_test)
    acc = accuracy_score(y_test, y_pred)
    f1 = f1_score(y_test, y_pred)
    prec = precision_score(y_test, y_pred)
    recall_val = recall_score(y_test, y_pred)
    print(f"=== {algo_cls.__name__} ===")
    print("--------------------------------")
    print(f"Accuracy: {round(acc * 100, 2)}%")
    print(f"F1 Score: {round(f1 * 100, 2)}%")
    print(f"Precision: {round(prec * 100, 2)}%")
    print(f"Recall: {round(recall_val * 100, 2)}%")
    print("--------------------------------")
    return (algo_cls.__name__, acc, f1, prec, recall_val)

if __name__ == '__main__':
    from sklearn.naive_bayes import BernoulliNB
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.svm import SVC
    from sklearn.linear_model import SGDClassifier
    from sklearn.preprocessing import StandardScaler

    start = time.perf_counter()

    # 数据加载与预处理
    df = pd.read_csv(".\\DATA\\heart.csv")
    X = df.iloc[:, :-1]
    y = df["output"]  # 使用Series而非DataFrame
    x_train, x_test, y_train, y_test = train_test_split(X, y, test_size=0.15, random_state=53)
    
    scaler = StandardScaler()
    x_train = scaler.fit_transform(x_train)
    x_test = scaler.transform(x_test)

    # 算法列表
    algo_classes = [BernoulliNB, RandomForestClassifier, SVC, SGDClassifier]

    # 多进程执行
    with concurrent.futures.ProcessPoolExecutor() as executor:
        # 提交任务，传入所有必要参数
        futures = [executor.submit(train, cls, x_train, x_test, y_train, y_test) for cls in algo_classes]
        # 收集结果（可选）
        for future in concurrent.futures.as_completed(futures):
            try:
                result = future.result()
            except Exception as e:
                print(f"训练出错: {e}")

    end = time.perf_counter()
    print(f'程序运行时间: {round((end - start) * 1000, 2)} ms')

关于ThreadPoolExecutor无性能提升的说明 #

Sklearn的大部分模型训练是CPU密集型任务，而Python的GIL（全局解释器锁）会限制多线程的CPU并行能力，所以ThreadPoolExecutor无法提升CPU密集型任务的运行速度，必须用ProcessPoolExecutor才能利用多核CPU。

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