Comment paralléliser apply () sur les DataFrames de Pandas en utilisant tous les noyaux sur une seule machine?

Le plus simple est d'utiliser Dask de map_partitions. Vous avez besoin de ces importations (vous aurez besoin d' pip install dask):

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

et la syntaxe est

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)  

(je crois que 30 est un nombre convenable de cloisons si vous avez 16 noyaux). Juste pour être complet, j'ai programmé la différence sur ma machine (16 cores):

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))

0.010668013244867325

Donner facteur de 10 speedup aller de pandas appliquer à dask appliquer sur les cloisons. Bien sûr, si vous avez une fonction que vous pouvez vectoriser, vous devriez - dans ce cas la fonction (y*(x**2+1)) est trivialement vectorisé, mais il ya beaucoup de choses qui sont impossibles à vectoriser.