python - 如何在 Python 中并行处理 pandas DataFrame 的行

Question

示例虚拟示例:

我有一个 DataFrame df:

> df

       para0  para1   para2
0  17.439020   True    high
1  19.757758   True    high
2  12.434424   True  medium
3  14.789654   True     low
4  14.131464  False    high
5   9.900233   True    high
6  10.977869  False     low
7   8.004251   True  medium
8  11.468420  False     low
9  12.764453  False    high

其中每一行由函数的参数集合组成foobar:

def foobar(r):
    """ r is a row of df, does something, and it takes a long time"""
    if r.para1:
        x = r.para2
    else:
        x = 'low'
    return int(r.para0), (r.Index+13)%3 == 0, x

我想将 foobar 应用于 df 的每一行，收集其结果，并将这些结果与其相关参数一起存储在，好吧，DataFrame。

我的(当前)解决方案:

df['count'] = 0
df['valid'] = False
df['outpt'] = ''

def wrapper(r, df):
    c, v, o = foobar(r)
    df.ix[r.Index,'count'] = c
    df.ix[r.Index,'valid'] = v
    df.ix[r.Index,'outpt'] = o

for r in df.itertuples():
    wrapper(r, df)

这会产生:

> df
       para0  para1   para2  count  valid   outpt
0  17.439020   True    high   17.0  False    high
1  19.757758   True    high   19.0  False    high
2  12.434424   True  medium   12.0   True  medium
3  14.789654   True     low   14.0  False     low
4  14.131464  False    high   14.0  False     low
5   9.900233   True    high    9.0   True    high
6  10.977869  False     low   10.0  False     low
7   8.004251   True  medium    8.0  False  medium
8  11.468420  False     low   11.0   True     low
9  12.764453  False    high   12.0  False     low

这是我的问题:

在现实生活中，函数 foobar 是计算成本较高，运行大约需要 20-30 分钟，df通常有 100-2000 行。我可以访问一台机器八个核心，并且 foobar 仅取决于当前处理的行无论是在其他任何事情上，运行这些都应该是微不足道的并行计算。

如果出现问题(例如，如果有人意外关闭机器)，无需启动一切从头开始，即跳过已经存在的行已处理。

我该怎么做？

<小时/>

不幸的是，我对多处理的尝试失败了:

from multiprocessing import Pool

pool = Pool(3)
results = []

for r in df.itertuples():
    results += [pool.apply_async(wrapper, r, df)]

与:

> results[0].get()
…
/usr/lib/python3.5/multiprocessing/reduction.py in dumps(cls, obj, protocol)
     48     def dumps(cls, obj, protocol=None):
     49         buf = io.BytesIO()
---> 50         cls(buf, protocol).dump(obj)
     51         return buf.getbuffer()
     52

PicklingError: Can't pickle <class 'pandas.core.frame.Pandas'>: attribute lookup Pandas on pandas.core.frame failed

以下是我创建玩具 DataFrame 的方法:

import pandas as pd
import numpy as np

df = pd.DataFrame({
    'para0' : pd.Series(
        np.random.gamma(12,size=10),
        dtype=np.float),
    'para1' : pd.Series(
        [(True,False)[i] for i in np.random.randint(0,2,10)],
        dtype=np.bool),
    'para2' : pd.Categorical(
        [('low','medium','high')[i] for i in np.random.randint(0,3,10)],
        ordered=True),
    })

Answer 1

我不知道这是否有帮助，但尝试使用list而不是itertuples。

我的意思是这样的:

df_list = [[x[0], x[1],x[2]] for x in df.itertuples()]
for r in df_list:
    results += [pool.apply_async(wrapper, r, df)]