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

Question

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

 "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

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

我们可以通过以下任一方法解决这个问题：

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

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