Faire en sorte que Pandas DataFrame apply() utilise tous les cœurs ?

Question

Depuis août 2017, les pandas DataFame.apply() est malheureusement toujours limité à un seul cœur, ce qui signifie qu'une machine à plusieurs cœurs gaspillera la majorité de son temps de calcul lorsque vous exécutez le programme df.apply(myfunc, axis=1) .

Comment utiliser tous vos cœurs pour exécuter une application sur un cadre de données en parallèle ?

Answer 1

Vous pouvez utiliser le swifter paquet :

pip install swifter

(Notez que vous pouvez vouloir utiliser ceci dans un virtualenv pour éviter les conflits de version avec les dépendances installées).

Swifter fonctionne comme un plugin pour pandas, ce qui vous permet de réutiliser la fonction apply fonction :

import swifter

def some_function(data):
    return data * 10

data['out'] = data['in'].swifter.apply(some_function)

Il trouvera automatiquement la manière la plus efficace de paralléliser la fonction, qu'elle soit vectorisée (comme dans l'exemple ci-dessus) ou non.

Autres exemples et un comparaison des performances sont disponibles sur GitHub. Notez que le paquet est en cours de développement actif, donc l'API peut changer.

Notez également que cette ne fonctionnera pas automatiquement pour les colonnes de type chaîne. Lors de l'utilisation de chaînes de caractères, Swifter se rabattra sur un "simple" Pandas apply qui ne seront pas parallèles. Dans ce cas, même en forçant l'utilisation de dask n'améliorera pas les performances, et vous feriez mieux de diviser votre ensemble de données manuellement et de paralléliser en utilisant multiprocessing .

Answer 2

Le moyen le plus simple est d'utiliser map_partitions de Dask . Vous avez besoin de ces importations (vous devrez 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 pense que 30 est un nombre approprié de partitions si vous avez 16 cœurs). Juste pour être complet, j'ai chronométré la différence sur ma machine (16 cœurs) :

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 un Accélération d'un facteur 10 passer de pandas apply à dask apply sur les partitions. 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 y a beaucoup de choses qui sont impossibles à vectoriser.

Answer 3

Vous pouvez essayer pandarallel à la place : Un outil simple et efficace pour paralléliser vos opérations pandas sur tous vos CPUs (Sur Linux & macOS)

• La parallélisation a un coût (instanciation de nouveaux processus, envoi de données via la mémoire partagée, etc ...), donc la parallélisation n'est efficace que si la quantité de calcul à paralléliser est suffisamment élevée. Pour une très petite quantité de données, utiliser la parallélisation ne vaut pas toujours la peine.

• Les fonctions appliquées ne doivent PAS être des fonctions lambda.

  from pandarallel import pandarallel from math import sin

  pandarallel.initialize()

  FORBIDDEN

  df.parallel_apply(lambda x: sin(x**2), axis=1)

  ALLOWED

  def func(x): return sin(x**2)

  df.parallel_apply(func, axis=1)

voir https://github.com/nalepae/pandarallel

Answer 4

Si vous voulez rester en python natif :

import multiprocessing as mp

with mp.Pool(mp.cpu_count()) as pool:
    df['newcol'] = pool.map(f, df['col'])

appliquera la fonction f de façon parallèle à la colonne col de l'image de données df

Answer 5

Voici un exemple de transformateur de base sklearn, dans lequel pandas apply est parallélisé

import multiprocessing as mp
from sklearn.base import TransformerMixin, BaseEstimator

class ParllelTransformer(BaseEstimator, TransformerMixin):
    def __init__(self,
                 n_jobs=1):
        """
        n_jobs - parallel jobs to run
        """
        self.variety = variety
        self.user_abbrevs = user_abbrevs
        self.n_jobs = n_jobs
    def fit(self, X, y=None):
        return self
    def transform(self, X, *_):
        X_copy = X.copy()
        cores = mp.cpu_count()
        partitions = 1

        if self.n_jobs <= -1:
            partitions = cores
        elif self.n_jobs <= 0:
            partitions = 1
        else:
            partitions = min(self.n_jobs, cores)

        if partitions == 1:
            # transform sequentially
            return X_copy.apply(self._transform_one)

        # splitting data into batches
        data_split = np.array_split(X_copy, partitions)

        pool = mp.Pool(cores)

        # Here reduce function - concationation of transformed batches
        data = pd.concat(
            pool.map(self._preprocess_part, data_split)
        )

        pool.close()
        pool.join()
        return data
    def _transform_part(self, df_part):
        return df_part.apply(self._transform_one)
    def _transform_one(self, line):
        # some kind of transformations here
        return line

Pour plus d'informations, voir https://towardsdatascience.com/4-easy-steps-to-improve-your-machine-learning-code-performance-88a0b0eeffa8