Python：pandas 数据帧上的多处理错误：客户端具有非平凡状态，即本地且不可腌制

Question

我有一个数据框，我使用groupby 将其划分为多个数据框。现在我想处理我编写了函数process_s2id并行的每个数据帧。我在class 中有整个代码，我正在使用另一个文件中的主函数执行该代码。但我收到以下错误：

 "Clients have non-trivial state that is local and unpickleable.",
_pickle.PicklingError: Pickling client objects is explicitly not supported.
Clients have non-trivial state that is local and unpickleable.

以下是代码（我们在这个类中执行main()函数）：

import logging
import pandas as pd
from functools import partial
from multiprocessing import Pool, cpu_count

class TestClass:
    
    def __init__(self):
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger()
        
    def process_s2id(self, df, col, new_col):
        dim2 = ['s2id', 'date', 'hours']
        df_hour = df.groupby(dim2)[[col, 'orders']].sum().reset_index()
        df_hour[new_col] = df_hour[col] / df_hour['orders']
        df_hour = df_hour[dim2 + [new_col]]
        return df_hour
    
    def run_parallel(self, df):
        series = [frame for keys, frame in df.groupby('s2id')]

        p = Pool(cpu_count())
        prod_x = partial(
            self.process_s2id,
            col ="total_supply",
            new_col = "supply"
        )
        s2id_supply_list = p.map(prod_x, series)
        p.close()
        p.join()

        s2id_supply = pd.concat(s2id_supply_list, axis=0)
        return ms2id_bsl
    
    def main(self):
        data = pd.read_csv("data/interim/fs.csv")
        out = self.run_parallel(data)
        return out

我尝试在 Spyder 中运行此代码，它运行良好。但是当我从另一个文件执行它时。我收到一个错误。以下是执行文件代码及错误：

import TestClass

def main():
    tc = TestClass()
    data = tc.main()

if __name__ == '__main__':
    main()

当我查看错误回溯时，我发现错误发生在函数开始并行的s2id_supply_list = p.map(prod_x, series) 行上。我也尝试连续运行它并且它有效。另外，我注意到这个特定的错误来自谷歌云包的client.py。有一个特定的代码，我在其中将数据上传到谷歌云，但该代码应该是不变的。我尝试努力搜索此错误，但所有结果都链接到 Google 云包相关问题，而不是多处理包。

谁能帮我理解这个错误，我该如何解决？

其他信息： 我有以下版本的软件包：

python==3.7.7
pandas==1.0.5
google-cloud-storage==1.20.0
google-cloud-core==1.0.3

我在 macbook pro 上运行这个。

Answer 1

我想通了。当我们在函数上使用Pool 来并行运行它时，它期望第一个参数是迭代器。换句话说，该函数将在第一个参数的不同值上并行运行。当我们在一个类中有一个非静态函数时，我们的第一个参数是self 或类本身。但是愚蠢的Pool 函数不知道如何使用self 进行迭代，因为它是错误的参数。正确的论点是第二个。

我们可以通过以下方式解决这个问题：

1. 从类中取出函数并将self 从参数中剔除。
2. 在函数顶部添加@staticmethod 并将self 从参数中踢出。

我希望这对遇到类似问题的人有所帮助。