Wie parallelisieren Sie apply() auf Pandas Dataframes, die alle Kerne auf einer Maschine nutzen?

Question

Ab August 2017 ist Pandas DataFame.apply() leider immer noch auf die Arbeit mit einem einzelnen Kern beschränkt, was bedeutet, dass ein Multi-Core-Rechner den größten Teil seiner Rechenzeit verschwendet, wenn Sie df.apply(myfunc, axis=1) ausführen.

Wie können Sie alle Ihre Kerne parallel auf einem Datenrahmen anwenden anwenden?

Answer 1

Der einfachste Weg ist Dask's map_partitions zu verwenden. Sie müssen diese Importe (Sie werden zu pip install dask benötigen):

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

und die Syntax ist

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) 

(Ich glaube, dass 30 eine geeignete Anzahl von Partitionen ist, wenn Sie 16 Kerne). Nur der Vollständigkeit halber, timed ich den Unterschied auf meinem Rechner (16 Kerne):

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

Giving eine Faktor 10 Speedup von Pandas gehen gelten für DASK auf Partitionen anwenden. Natürlich, wenn Sie eine Funktion haben, die Sie vektorisieren können, sollten Sie - in diesem Fall ist die Funktion (y*(x**2+1)) trivial vektorisiert, aber es gibt viele Dinge, die unmöglich zu vektorisieren sind.