我经常在超过1500万行左右的数据帧上执行pandas操作,我希望能够访问特定操作的进度指示器。

是否存在基于文本的熊猫分裂-应用-组合操作进度指示器?

例如:

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

其中feature_rollup是一个有点复杂的函数,它采用许多DF列,并通过各种方法创建新的用户列。对于大数据帧,这些操作可能需要一段时间,所以我想知道是否有可能在iPython笔记本中有基于文本的输出,以更新我的进度。

到目前为止,我已经尝试了Python的规范循环进度指示器,但它们没有以任何有意义的方式与pandas交互。

我希望在pandas库/文档中有一些我忽略了的东西,可以让人们了解分裂-应用-组合的进展。一个简单的实现可能会查看apply函数正在处理的数据帧子集的总数,并将进度报告为这些子集的完成部分。

这可能是需要添加到库中的东西吗?


当前回答

concat操作:

df = pd.concat(
    [
        get_data(f)
        for f in tqdm(files, total=len(files))
    ]
)

TQDM只返回一个可迭代对象。

其他回答

对于希望在自定义并行熊猫应用代码上应用tqdm的任何人。

(多年来,我尝试了一些并行化的库,但我从未找到100%的并行化解决方案,主要是针对apply函数,而且我总是不得不回来查看我的“手动”代码。)

Df_multi_core——这是你要调用的。它接受:

df对象 要调用的函数名 函数可以执行的列的子集(有助于减少时间/内存) 并行运行的作业数量(-1或忽略所有核心) df函数接受的任何其他kwargs(如“axis”)

_df_split——这是一个内部辅助函数,必须全局定位到正在运行的模块(Pool. split)。地图是“位置依赖”),否则我会在内部定位它..

以下是我的gist代码(我将在那里添加更多的pandas函数测试):

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

在输出中,您可以看到在没有并行化的情况下运行的进度条,以及在并行化运行时的每个核心进度条。 有一个轻微的hickup,有时其余的核心出现在一次,但即使这样,我认为它是有用的,因为你得到每个核心的进度统计(它/秒和总记录,为ex)

感谢@abcdaa提供这么棒的图书馆!

如果你像我一样需要在Jupyter/ipython笔记本中如何使用它,这里有一个有用的指南和相关文章的来源:

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)

请注意_tqdm_notebook的导入语句中的下划线。正如参考文章所提到的,开发正处于后期测试阶段。

截至2021年12月11日更新

我目前正在使用pandas==1.3.4和tqdm==4.62.3,我不确定tqdm作者是哪个版本实现了这一更改,但上面的import语句已弃用。而不是使用:

 from tqdm.notebook import tqdm_notebook

截至2022年1月2日更新 现在可以简化.py和.ipynb文件的导入语句:

from tqdm.auto import tqdm
tqdm.pandas()

这应该可以在两种开发环境中正常工作,并且应该在pandas数据框架或其他值得tqdm支持的可迭代对象上工作。

截至2022年5月27日更新 如果你在SageMaker上使用jupyter笔记本,这个组合是有效的:

from tqdm import tqdm
from tqdm.gui import tqdm as tqdm_gui
tqdm.pandas(ncols=50)

你可以很容易地用装饰器做到这一点

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)

然后使用modified_function(当你想打印它时更改)

对于合并、concat、join等操作,可以使用Dask显示进度条。

您可以将Pandas数据框架转换为Dask数据框架。然后你可以显示Dask进度条。

下面的代码显示了一个简单的例子:

创建和转换Pandas数据框架

import pandas as pd
import numpy as np
from tqdm import tqdm
import dask.dataframe as dd

n = 450000
maxa = 700

df1 = pd.DataFrame({'lkey': np.random.randint(0, maxa, n),'lvalue': np.random.randint(0,int(1e8),n)})
df2 = pd.DataFrame({'rkey': np.random.randint(0, maxa, n),'rvalue': np.random.randint(0, int(1e8),n)})

sd1 = dd.from_pandas(df1, npartitions=3)
sd2 = dd.from_pandas(df2, npartitions=3)

使用进度条合并

from tqdm.dask import TqdmCallback
from dask.diagnostics import ProgressBar
ProgressBar().register()

with TqdmCallback(desc="compute"):
    sd1.merge(sd2, left_on='lkey', right_on='rkey').compute()

同样的操作,Dask比Pandas更快,需要的资源更少:

熊猫74.7毫秒 Dask 20.2 ms

详情如下:

进度条合并或连接操作与tqdm在熊猫 测试笔记本

注1:我已经测试了这个解决方案:https://stackoverflow.com/a/56257514/3921758,但它不适合我。不度量合并操作。

注2:我已经为熊猫的tqdm检查了“开放请求”,如下:

https://github.com/tqdm/tqdm/issues/1144 https://github.com/noamraph/tqdm/issues/28

concat操作:

df = pd.concat(
    [
        get_data(f)
        for f in tqdm(files, total=len(files))
    ]
)

TQDM只返回一个可迭代对象。