1台のマシン上のすべてのコアを利用するPandas Dataframesでapply（）をどのように並列化しますか？

Question

2017年8月現在、残念ながら単コアでの作業に限定されています。つまり、df.apply(myfunc, axis=1)を実行すると、マルチコアマシンで計算時間の大半が無駄になります。

データフレームに適用するすべてのコアを並列でどのように使用できますか？

Answer 1

最も簡単な方法はDask's map_partitionsです。

import pandas as pd 
import dask.dataframe as dd 
from dask.multiprocessing import get 

をし、構文は

data = <your_pandas_dataframe> 
ddata = dd.from_pandas(data, npartitions=30) 

def myfunc(x,y,z, ...): return <whatever> 

res = ddata.map_partitions(lambda df: df.apply((lambda row: myfunc(*row)), axis=1)).compute(get=get) 

（私はあなたが16個のコアを持っている場合、30パーティションの適切な数であると信じている）である：あなたは（あなたがpip install daskにする必要があります）これらの輸入を必要としています。

data = pd.DataFrame() 
data['col1'] = np.random.normal(size = 1500000) 
data['col2'] = np.random.normal(size = 1500000) 

ddata = dd.from_pandas(data, npartitions=30) 
def myfunc(x,y): return y*(x**2+1) 
def apply_myfunc_to_DF(df): return df.apply((lambda row: myfunc(*row)), axis=1) 
def pandas_apply(): return apply_myfunc_to_DF(data) 
def dask_apply(): return ddata.map_partitions(apply_myfunc_to_DF).compute(get=get) 
def vectorized(): return myfunc(data['col1'], data['col2'] ) 

t_pds = timeit.Timer(lambda: pandas_apply()) 
print(t_pds.timeit(number=1)) 

28.16970546543598

t_dsk = timeit.Timer(lambda: dask_apply()) 
print(t_dsk.timeit(number=1)) 

2.708152851089835

t_vec = timeit.Timer(lambda: vectorized()) 
print(t_vec.timeit(number=1)) 

：ちょうど完全を期すために、私は私のマシン上の違い（16コア）を時限はパンダから行く10のスピードアップの要因 を与える

0.010668013244867325

はパーティションに適用DASKに適用されます。もちろん、ベクトル化できる関数がある場合は、関数（y*(x**2+1)）は単純にベクトル化されますが、ベクトル化することは不可能なことがたくさんあります。