pandas 并行分组

Question

我在跨多个核心拆分分组操作的聚合步骤时遇到了一些问题。我有以下工作代码，并想将其应用于多个处理器：

import pandas as pd
import numpy as np
from multiprocessing import Pool, cpu_count

mydf = pd.DataFrame({'v1':[1,2,3,4]*6,'v2':['a','b','c']*8,'v3':np.arange(20,44)})

然后我可以应用以下 GroupBy 操作：（我希望并行执行的步骤）

pd.groupby(mydf,by=['v1','v2']).apply(lambda x: np.percentile(x['v3'],[20,30]))

产生系列：

1   a     [22.4, 23.6]
    b     [26.4, 27.6]
    c     [30.4, 31.6]
2   a     [31.4, 32.6]
    b     [23.4, 24.6]
    c     [27.4, 28.6]

我尝试了以下，参考：parallel groupby

def applyParallel(dfGrouped, func):
    with Pool(1) as p:
        ret_list = p.map(func, [group for name, group in dfGrouped])
    return pd.concat(ret_list)

def myfunc(df):
    df['pct1'] = df.loc[:,['v3']].apply(np.percentile,args=([20],))
    df['pct2'] = df.loc[:,['v3']].apply(np.percentile,args=([80],))
    return(df)


grouped = pd.groupby(mydf,by=['v1','v2'])
applyParallel(grouped,myfunc)

但我正在丢失索引结构并得到重复项。我可能可以通过操作进一步分组来解决这一步，但我认为完全避免它应该不会太难。有什么建议吗？

Answer 1

并不是说我仍在寻找答案，但最好使用处理 pandas DataFrame 并行操作的库，而不是尝试手动操作。

Dask 是一个选项，旨在通过很少的代码修改来扩展 Pandas 操作。

另一个选项（但设置起来可能有点困难）是 PySpark

pandas 并行分组

pandas group by in parallel

parallel-processing

python-3.x

pandas

pandas-groupby