Pandas DataFrame并行化操作代码挂起问题求解

原代码的核心问题 #

1. 未传递拆分后的DataFrame：循环遍历splits但在apply_async里没把split作为参数传入，工作进程没有处理目标数据。
2. 方法引用错误：直接使用operation而不是self.operation，子进程无法定位该方法；Windows环境下还会触发序列化失败问题。
3. 参数格式错误：args=(2)不是合法元组（单元素元组需加逗号），且缺少operation要求的第一个参数df。
4. 类方法的多进程兼容性：Windows系统中multiprocessing默认用spawn模式，类实例方法无法直接被子进程序列化调用。

修复后的代码实现 #

方案1：提取操作为顶层函数（兼容性最优） #

import numpy as np
import multiprocessing as mp
import pandas as pd

# 把操作提取为顶层函数，规避类方法序列化问题
def operation(df, param1):
    # 示例操作：生成包含自定义类的字典
    class ResultItem:
        def __init__(self, value):
            self.value = value
    return {idx: ResultItem(row.sum()) for idx, row in df.iterrows()}

class DFOperator:
    def __init__(self, df, num_cores):
        self.num_cores = num_cores
        self.df = df

    def task(self):
        splits = np.array_split(self.df, self.num_cores)
        # 适配Windows环境的spawn启动模式（Linux/macOS可忽略）
        mp.set_start_method('spawn', force=True)
        with mp.Pool(self.num_cores) as p:
            # 正确传递拆分后的df和参数，args需为元组格式
            async_results = [p.apply_async(operation, args=(split, 2)) for split in splits]
            # 收集并合并结果
            results = []
            for ar in async_results:
                results.extend(ar.get().values())
            return results

方案2：保留类方法（仅适用于Linux/macOS） #

Linux/macOS默认用fork模式，可直接调用类方法，只需修正参数传递：

import numpy as np
import multiprocessing as mp
import pandas as pd

class DFOperator:
    def __init__(self, df, num_cores):
        self.num_cores = num_cores
        self.df = df

    def operation(self, df, param1):
        class ResultItem:
            def __init__(self, value):
                self.value = value
        return {idx: ResultItem(row.sum()) for idx, row in df.iterrows()}

    def task(self):
        splits = np.array_split(self.df, self.num_cores)
        with mp.Pool(self.num_cores) as p:
            # 改用self.operation，并传入拆分后的df
            async_results = [p.apply_async(self.operation, args=(split, 2)) for split in splits]
            results = []
            for ar in async_results:
                results.extend(ar.get().values())
            return results

额外注意事项 #

• 序列化问题：如果自定义类无法被序列化，可添加__reduce__方法，或改用字典等原生可序列化结构替代。
• 成本权衡：若DataFrame体量小，多进程的启动和数据传递开销可能超过并行收益，建议单进程运行。
• 进程数设置：num_cores建议设为mp.cpu_count()或其减1，避免占用全部CPU资源。

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