Indicador de progresso durante operações de pandas

Realizo regularmente operações de pandas em quadros de dados com mais de 15 milhões de linhas e gostaria de ter acesso a um indicador de progresso para operações específicas.

Existe um indicador de progresso baseado em texto para operações de divisão de aplicação e combinação de pandas?

Por exemplo, em algo como:

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

onde feature_rollupé uma função um pouco envolvida que pega muitas colunas do DF e cria novas colunas de usuário por meio de vários métodos. Essas operações podem demorar um pouco para grandes quadros de dados, então eu gostaria de saber se é possível ter uma saída baseada em texto em um notebook iPython que me atualize sobre o progresso.

Até agora, tentei indicadores de progresso de loop canônico para Python, mas eles não interagem com os pandas de maneira significativa.

Espero que haja algo que eu tenha esquecido na biblioteca / documentação do pandas que permita conhecer o progresso de uma combinação de aplicar e dividir. Uma implementação simples talvez analise o número total de subconjuntos de quadros de dados nos quais a applyfunção está funcionando e relate o progresso como a fração concluída desses subconjuntos.

Talvez isso seja algo que precise ser adicionado à biblioteca?

Devido à demanda popular, tqdmadicionou suporte para pandas. Diferentemente das outras respostas, isso não deixará os pandas mais lentos para baixo - eis um exemplo para 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)

Caso você esteja interessado em saber como isso funciona (e como modificá-lo para seus próprios retornos de chamada), veja os exemplos no github , a documentação completa sobre o pypi ou importe o módulo e execute-o help(tqdm).

EDITAR

Para responder diretamente à pergunta original, substitua:

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

com:

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

Nota: tqdm <= v4.8 : Para versões do tqdm abaixo de 4.8, em vez de tqdm.pandas()você precisar:

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

Ajustar a resposta de Jeff (e ter isso como uma função reutilizável).

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

Nota: aplique as atualizações de porcentagem de progresso em linha . Se sua função persistir, isso não funcionará.

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

In [12]: f = feature_rollup

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

Como de costume, você pode adicionar isso ao seu grupo por objetos como um método:

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

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

Como mencionado nos comentários, esse não é um recurso que os principais pandas estariam interessados ​​em implementar. Mas o python permite que você os crie para muitos objetos / métodos do pandas (isso seria um pouco de trabalho ... embora você deva ser capaz de generalizar essa abordagem).

Caso você precise de suporte para como usar isso em um notebook Jupyter / ipython, como eu fiz, aqui está um guia e uma fonte úteis para o artigo relevante :

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)

Observe o sublinhado na instrução de importação para _tqdm_notebook. Como o artigo referenciado menciona, o desenvolvimento está no estágio beta tardio.

Para quem deseja aplicar o tqdm em seu código de aplicação de pandas paralelos personalizado.

(Tentei algumas das bibliotecas para paralelização ao longo dos anos, mas nunca encontrei uma solução 100% de paralelização, principalmente para a função de aplicação, e sempre tive que voltar para o meu código "manual".)

df_multi_core - este é o que você chama. Aceita:

1. Seu objeto df
2. O nome da função que você deseja chamar
3. O subconjunto de colunas em que a função pode ser executada (ajuda a reduzir o tempo / memória)
4. O número de tarefas a serem executadas em paralelo (-1 ou omitir para todos os núcleos)
5. Quaisquer outros kwargs aceitos pela função do df (como "eixo")

_df_split - esta é uma função auxiliar interna que deve ser posicionada globalmente para o módulo em execução (Pool.map é "dependente de posicionamento"), caso contrário, eu a localizaria internamente.

aqui está o código da minha essência (vou adicionar mais testes de função de pandas lá):

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

Abaixo está um código de teste para uma aplicação paralela com 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))

Na saída, você pode ver 1 barra de progresso para execução sem paralelização e barras de progresso por núcleo ao executar com paralelização. Há um pequeno hickup e, às vezes, o restante dos núcleos aparece de uma só vez, mas mesmo assim acho útil, pois você obtém as estatísticas de progresso por núcleo (ele / s e total de registros, por ex)

Obrigado @abcdaa por esta ótima biblioteca!

Você pode fazer isso facilmente com um decorador

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)

basta usar a função modified_ (e altere quando quiser imprimir)

Eu mudei a resposta de Jeff , para incluir um total, para que você possa acompanhar o progresso e uma variável para apenas imprimir cada X iterações (isso realmente melhora o desempenho por um monte, se o "print_at" é razoavelmente elevado)

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

a função clear_output () é de

from IPython.core.display import clear_output

se não no IPython, a resposta de Andy Hayden faz isso sem ele