Indicateur de progression pendant les opérations pandas

J'effectue régulièrement des opérations de pandas sur des trames de données de plus de 15 millions de lignes et j'aimerais avoir accès à un indicateur de progression pour des opérations particulières.

Existe-t-il un indicateur de progression basé sur du texte pour les opérations de fractionnement-appliquer-combiner pandas?

Par exemple, dans quelque chose comme:

df_users.groupby(['userID', 'requestDate']).apply(feature_rollup)

où feature_rollupest une fonction quelque peu impliquée qui prend de nombreuses colonnes DF et crée de nouvelles colonnes utilisateur par diverses méthodes. Ces opérations peuvent prendre un certain temps pour les grandes trames de données, j'aimerais donc savoir s'il est possible d'avoir une sortie texte dans un bloc-notes iPython qui me met à jour sur la progression.

Jusqu'à présent, j'ai essayé des indicateurs de progression de boucle canoniques pour Python, mais ils n'interagissent pas avec les pandas de manière significative.

J'espère qu'il y a quelque chose que j'ai négligé dans la bibliothèque / documentation de pandas qui permet de connaître la progression d'un split-apply-combine. Une implémentation simple pourrait peut-être examiner le nombre total de sous-ensembles de trames de données sur lesquels la applyfonction travaille et rapporter la progression comme la fraction terminée de ces sous-ensembles.

Est-ce peut-être quelque chose qui doit être ajouté à la bibliothèque?

En raison de la demande populaire, tqdma ajouté un support pour pandas. Contrairement aux autres réponses, cela ne ralentira pas sensiblement les pandas - voici un exemple pour DataFrameGroupBy.progress_apply:

import pandas as pd
import numpy as np
from tqdm import tqdm
# from tqdm.auto import tqdm  # for notebooks

df = pd.DataFrame(np.random.randint(0, int(1e8), (10000, 1000)))

# Create and register a new `tqdm` instance with `pandas`
# (can use tqdm_gui, optional kwargs, etc.)
tqdm.pandas()

# Now you can use `progress_apply` instead of `apply`
df.groupby(0).progress_apply(lambda x: x**2)

Si vous souhaitez savoir comment cela fonctionne (et comment le modifier pour vos propres rappels), consultez les exemples sur github , la documentation complète sur pypi , ou importez le module et exécutez help(tqdm).

ÉDITER

Pour répondre directement à la question d'origine, remplacez:

df_users.groupby(['userID', 'requestDate']).apply(feature_rollup)

avec:

from tqdm import tqdm
tqdm.pandas()
df_users.groupby(['userID', 'requestDate']).progress_apply(feature_rollup)

Remarque: tqdm <= v4.8 : Pour les versions de tqdm inférieures à 4.8, au lieu de cela, tqdm.pandas()vous deviez faire:

from tqdm import tqdm, tqdm_pandas
tqdm_pandas(tqdm())

Pour modifier la réponse de Jeff (et avoir ceci comme fonction réutilisable).

def logged_apply(g, func, *args, **kwargs):
    step_percentage = 100. / len(g)
    import sys
    sys.stdout.write('apply progress:   0%')
    sys.stdout.flush()

    def logging_decorator(func):
        def wrapper(*args, **kwargs):
            progress = wrapper.count * step_percentage
            sys.stdout.write('\033[D \033[D' * 4 + format(progress, '3.0f') + '%')
            sys.stdout.flush()
            wrapper.count += 1
            return func(*args, **kwargs)
        wrapper.count = 0
        return wrapper

    logged_func = logging_decorator(func)
    res = g.apply(logged_func, *args, **kwargs)
    sys.stdout.write('\033[D \033[D' * 4 + format(100., '3.0f') + '%' + '\n')
    sys.stdout.flush()
    return res

Note: le pourcentage appliquent progrès mises à jour en ligne . Si votre fonction stdouts, cela ne fonctionnera pas.

In [11]: g = df_users.groupby(['userID', 'requestDate'])

In [12]: f = feature_rollup

In [13]: logged_apply(g, f)
apply progress: 100%
Out[13]: 
...

Comme d'habitude, vous pouvez ajouter ceci à vos objets groupby en tant que méthode:

from pandas.core.groupby import DataFrameGroupBy
DataFrameGroupBy.logged_apply = logged_apply

In [21]: g.logged_apply(f)
apply progress: 100%
Out[21]: 
...

Comme mentionné dans les commentaires, ce n'est pas une fonctionnalité que les pandas de base seraient intéressés à implémenter. Mais python vous permet de les créer pour de nombreux objets / méthodes pandas (cela serait un peu de travail ... bien que vous devriez être en mesure de généraliser cette approche).

Au cas où vous auriez besoin d'aide pour savoir comment l'utiliser dans un notebook Jupyter / ipython, comme je l'ai fait, voici un guide utile et une source vers l' article pertinent :

from tqdm._tqdm_notebook import tqdm_notebook
import pandas as pd
tqdm_notebook.pandas()
df = pd.DataFrame(np.random.randint(0, int(1e8), (10000, 1000)))
df.groupby(0).progress_apply(lambda x: x**2)

Notez le trait de soulignement dans l'instruction d'importation pour _tqdm_notebook. Comme le mentionne l'article référencé, le développement est en phase bêta tardive.

Pour tous ceux qui cherchent à appliquer tqdm sur leur code d'application de pandas parallèle personnalisé.

(J'ai essayé certaines des bibliothèques de parallélisation au fil des ans, mais je n'ai jamais trouvé de solution de parallélisation à 100%, principalement pour la fonction apply, et je devais toujours revenir pour mon code "manuel".)

df_multi_core - c'est celui que vous appelez. Il accepte:

1. Votre objet df
2. Le nom de la fonction que vous souhaitez appeler
3. Le sous-ensemble de colonnes sur lequel la fonction peut être exécutée (aide à réduire le temps / la mémoire)
4. Le nombre de travaux à exécuter en parallèle (-1 ou omis pour tous les cœurs)
5. Tout autre kwargs accepté par la fonction df (comme "axis")

_df_split - il s'agit d'une fonction d'assistance interne qui doit être positionnée globalement sur le module en cours d'exécution (Pool.map est "dépendant du placement"), sinon je la localiserais en interne ..

voici le code de mon résumé (j'ajouterai plus de tests de fonction pandas ici):

import pandas as pd
import numpy as np
import multiprocessing
from functools import partial

def _df_split(tup_arg, **kwargs):
    split_ind, df_split, df_f_name = tup_arg
    return (split_ind, getattr(df_split, df_f_name)(**kwargs))

def df_multi_core(df, df_f_name, subset=None, njobs=-1, **kwargs):
    if njobs == -1:
        njobs = multiprocessing.cpu_count()
    pool = multiprocessing.Pool(processes=njobs)

    try:
        splits = np.array_split(df[subset], njobs)
    except ValueError:
        splits = np.array_split(df, njobs)

    pool_data = [(split_ind, df_split, df_f_name) for split_ind, df_split in enumerate(splits)]
    results = pool.map(partial(_df_split, **kwargs), pool_data)
    pool.close()
    pool.join()
    results = sorted(results, key=lambda x:x[0])
    results = pd.concat([split[1] for split in results])
    return results

Ci-dessous est un code de test pour une application parallélisée avec tqdm "progress_apply".

from time import time
from tqdm import tqdm
tqdm.pandas()

if __name__ == '__main__': 
    sep = '-' * 50

    # tqdm progress_apply test      
    def apply_f(row):
        return row['c1'] + 0.1
    N = 1000000
    np.random.seed(0)
    df = pd.DataFrame({'c1': np.arange(N), 'c2': np.arange(N)})

    print('testing pandas apply on {}\n{}'.format(df.shape, sep))
    t1 = time()
    res = df.progress_apply(apply_f, axis=1)
    t2 = time()
    print('result random sample\n{}'.format(res.sample(n=3, random_state=0)))
    print('time for native implementation {}\n{}'.format(round(t2 - t1, 2), sep))

    t3 = time()
    # res = df_multi_core(df=df, df_f_name='apply', subset=['c1'], njobs=-1, func=apply_f, axis=1)
    res = df_multi_core(df=df, df_f_name='progress_apply', subset=['c1'], njobs=-1, func=apply_f, axis=1)
    t4 = time()
    print('result random sample\n{}'.format(res.sample(n=3, random_state=0)))
    print('time for multi core implementation {}\n{}'.format(round(t4 - t3, 2), sep))

Dans la sortie, vous pouvez voir 1 barre de progression pour l'exécution sans parallélisation et des barres de progression par cœur lors de l'exécution avec parallélisation. Il y a un léger accrochage et parfois le reste des cœurs apparaît à la fois, mais même dans ce cas, je pense que c'est utile car vous obtenez les statistiques de progression par cœur (it / sec et total des enregistrements, par exemple)

Merci @abcdaa pour cette superbe bibliothèque!

Vous pouvez facilement le faire avec un décorateur

from functools import wraps 

def logging_decorator(func):

    @wraps
    def wrapper(*args, **kwargs):
        wrapper.count += 1
        print "The function I modify has been called {0} times(s).".format(
              wrapper.count)
        func(*args, **kwargs)
    wrapper.count = 0
    return wrapper

modified_function = logging_decorator(feature_rollup)

puis utilisez simplement la fonction_modifiée (et changez quand vous voulez qu'elle s'imprime)

J'ai changé la réponse de Jeff , pour inclure un total, afin que vous puissiez suivre la progression et une variable pour simplement imprimer toutes les itérations X (cela améliore en fait beaucoup les performances, si le "print_at" est raisonnablement élevé)

def count_wrapper(func,total, print_at):

    def wrapper(*args):
        wrapper.count += 1
        if wrapper.count % wrapper.print_at == 0:
            clear_output()
            sys.stdout.write( "%d / %d"%(calc_time.count,calc_time.total) )
            sys.stdout.flush()
        return func(*args)
    wrapper.count = 0
    wrapper.total = total
    wrapper.print_at = print_at

    return wrapper

la fonction clear_output () est de

from IPython.core.display import clear_output

sinon sur IPython La réponse d'Andy Hayden le fait sans elle