[python] 팬더 작업 중 진행률 표시기

사용자 정의 병렬 팬더 적용 코드에 tqdm을 적용하려는 모든 사람에게 적합합니다.

(수년 동안 병렬화를 위해 라이브러리 중 일부를 시도했지만 주로 적용 기능에 대한 100 % 병렬화 솔루션을 찾지 못했으며 항상 “수동”코드로 돌아와야했습니다.

df_multi_core- 이것이 당신이 부르는 것입니다. 받아들입니다 :

1. df 객체
2. 호출하려는 함수 이름
3. 함수를 수행 할 수있는 열의 하위 집합 (시간 / 메모리 감소에 도움이 됨)
4. 병렬로 실행할 작업 수 (-1 또는 모든 코어에서 생략)
5. df의 함수가 받아들이는 다른 kwargs (예 : “axis”)

_df_split- 실행중인 모듈 (Pool.map은 “배치에 따라 다름”)에 전역으로 배치해야하는 내부 도우미 함수입니다. 그렇지 않으면 내부적으로 찾습니다 ..

여기 내 요지 코드가 있습니다 (팬더 기능 테스트를 더 추가 할 것입니다).

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

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

출력에서 병렬화없이 실행하기위한 진행률 표시 줄 1 개와 병렬화로 실행할 때 코어 당 진행률 표시 줄을 볼 수 있습니다. 약간의 hickup이 있고 때로는 나머지 코어가 한 번에 나타나지만 코어 당 진행 통계 (예 : 초당 / 총 레코드 수)를 얻었으므로 유용하다고 생각합니다.

이 훌륭한 도서관에 대해 @abcdaa에게 감사합니다!

답변

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)

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

from IPython.core.display import clear_output