我经常在超过1500万行的数据框上执行pandas操作，我希望能够在特定操作中获得进度指示器。
是否存在适用于pandas split-apply-combine操作的文本进度指示器？
例如，在以下操作中：
df_users.groupby(['userID', 'requestDate']).apply(feature_rollup)
在这里，feature_rollup是一个相当复杂的函数，它通过各种方法使用许多DF列创建新的用户列。对于大型数据帧，这些操作可能需要一段时间，因此我想知道是否可以在iPython笔记本中有基于文本的输出，以便更新进度。
到目前为止，我已经尝试了Python的常规循环进度指示器，但它们没有与Pandas进行任何有意义的交互。
我希望我在Pandas库/文档中忽略了一些东西，可以让人们知道拆分-应用-组合的进度。一个简单的实现可能会查看正在处理apply函数的数据帧子集的总数，并报告完成这些子集的分数。
这可能是需要添加到库中的内容吗？
应广大用户要求，我已在 tqdm 中添加了对 pandas 的支持（pip install "tqdm>=4.9.0"）。与其他答案不同，这不会明显减慢 pandas 的运行速度 -- 下面是一个使用 DataFrameGroupBy.progress_apply 的示例：
import pandas as pd
import numpy as np
from tqdm import tqdm
# from tqdm.auto import tqdm # for notebooks
# Create new `pandas` methods which use `tqdm` progress
# (can use tqdm_gui, optional kwargs, etc.)
tqdm.pandas()
df = pd.DataFrame(np.random.randint(0, int(1e8), (10000, 1000)))
# Now you can use `progress_apply` instead of `apply`
df.groupby(0).progress_apply(lambda x: x**2)
如果你对这个工作原理感兴趣（以及如何修改为自己的回调函数），请查看GitHub上的示例，在PyPI上查看完整文档，或导入模块并运行help(tqdm)。其他支持的函数包括map、applymap、aggregate和transform。
编辑
直接回答原问题，请将以下内容替换：
df_users.groupby(['userID', 'requestDate']).apply(feature_rollup)
跟随着：
from tqdm import tqdm
tqdm.pandas()
df_users.groupby(['userID', 'requestDate']).progress_apply(feature_rollup)
注意：tqdm <= v4.8：
对于版本小于4.8的tqdm，您需要执行以下操作而不是tqdm.pandas()：
tqdm.pandas()：from tqdm import tqdm, tqdm_pandas
tqdm_pandas(tqdm())
tqdm 最初是为普通的可迭代对象创建的：from tqdm import tqdm; for i in tqdm( range(int(1e8)) ): pass，pandas 的支持只是我最近添加的一个小技巧 :) - casper.dclfrom tqdm import tqdm_notebook; tqdm_notebook().pandas(*args, **kwargs)。请看这里。 - grinsbaeckchentqdm>=4.57.0中消失。 - casper.dcl如果您需要在Jupyter/ipython笔记本中使用此工具的支持，就像我一样，这里有一个有用的指南和相关来源 relevant article：
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年11月12日更新：
我目前正在使用pandas == 1.3.4 和 tqdm == 4.62.3，不确定tqdm作者是从哪个版本开始实施了这个更改，但上述导入语句已经被弃用。请使用以下导入语句：
from tqdm.notebook import tqdm_notebook
截至2022年02月01日更新 现在可以简化导入.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)
To tweak Jeff's answer (and have this as a reuseable function).
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
注意：应用程序进度百分比会在行内更新。如果您的函数有标准输出，则此方法不起作用。
In [11]: g = df_users.groupby(['userID', 'requestDate'])
In [12]: f = feature_rollup
In [13]: logged_apply(g, f)
apply progress: 100%
Out[13]:
...
像往常一样，您可以将此添加到您的groupBy对象作为一个方法：
from pandas.core.groupby import DataFrameGroupBy
DataFrameGroupBy.logged_apply = logged_apply
In [21]: g.logged_apply(f)
apply progress: 100%
Out[21]:
...
正如评论中提到的那样，这并不是核心pandas感兴趣的实现功能。但是Python允许您为许多pandas对象/方法创建这些（这样做需要相当大量的工作...尽管您应该能够将此方法概括）。
如果有人想在他们的自定义并行 pandas-apply 代码上应用 tqdm，以下是一些帮助：
多年来，我尝试过一些并行化库，但我从未找到100%的并行化解决方案，主要是应用函数，我总是不得不回到我的"手动"代码。
df_multi_core - 你需要调用它。它接受：
_df_split - 这是一个内部辅助函数，必须全局定位到运行模块(Pool.map 是"放置依赖项"的)，否则我会将其定位为内部函数。
以下是来自我的 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
以下是使用tqdm "progress_apply"并行化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))
try: splits = np.array_split(df[subset], njobs) except ValueError: splits = np.array_split(df, njobs)因为出现了KeyError异常而不是ValueError异常，所以改为使用Exception来处理所有情况。 - Mariuspandas.DataFrame.groupby。如果你想在没有groupby的情况下在pandas.Series.apply中使用进度条，以下是如何在jupyter-notebook中实现的：from tqdm.notebook import tqdm
tqdm.pandas()
df['<applied-col-name>'] = df['<col-name>'].progress_apply(<your-manipulation-function>)
tqdm 版本:tqdm>=4.61.2)，否则它不会工作。另外，在安装了新版本的 tqdm 之后，请确保重新启动你的 jupyternotebook 内核。(例如，我使用了 pip install tqdm==4.62.3) - Dr Neofrom 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)
logged_apply(g, func)函数中，这样您将可以访问顺序，并从一开始就进行日志记录。 - Andy Haydendef 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
clear_output()函数来自于
from IPython.core.display import clear_output
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毫秒20.2毫秒更多细节请参见:
注意1：我已经测试了这个解决方案：https://dev59.com/7bTma4cB1Zd3GeqPzwj-#56257514，但对我无效。不能测量合并操作。
注意2：我已经检查了Pandas中“tqdm”的“open request”：
关于连接操作：
df = pd.concat(
[
get_data(f)
for f in tqdm(files, total=len(files))
]
)
tqdm 只返回一个可迭代对象。
如果你想要遍历群组，这个方法可以帮到你
from tqdm import tqdm
groups = df.groupby(group_cols)
for keys, grouped_df in tqdm(groups, total=groups.ngroups)
pass