是否有一种简单的方法来跟踪 joblib.Parallel 执行的整体进度？
我有一个由数千个作业组成的长时间运行的执行，我想在数据库中对其进行跟踪和记录。但是，要做到这一点，每当 Parallel 完成任务时，我需要它执行回调，报告剩余的作业数。
我之前使用 Python 的 stdlib multiprocessing.Pool 完成了类似的任务，方法是启动一个线程来记录 Pool 的作业列表中待处理作业的数量。
看代码，Parallel继承了Pool，所以我想我可以实现同样的技巧，但它似乎没有使用这些列表，我一直无法弄清楚如何“阅读” " 任何其他方式都是内部状态。
【问题讨论】：
标签： python multithreading parallel-processing multiprocessing joblib
比 dano 和 Connor 的答案更进一步的是将整个事情包装为上下文管理器：
import contextlib
import joblib
from tqdm import tqdm
@contextlib.contextmanager
def tqdm_joblib(tqdm_object):
"""Context manager to patch joblib to report into tqdm progress bar given as argument"""
class TqdmBatchCompletionCallback(joblib.parallel.BatchCompletionCallBack):
def __call__(self, *args, **kwargs):
tqdm_object.update(n=self.batch_size)
return super().__call__(*args, **kwargs)
old_batch_callback = joblib.parallel.BatchCompletionCallBack
joblib.parallel.BatchCompletionCallBack = TqdmBatchCompletionCallback
try:
yield tqdm_object
finally:
joblib.parallel.BatchCompletionCallBack = old_batch_callback
tqdm_object.close()
然后你可以像这样使用它，完成后不要留下猴子修补的代码：
from joblib import Parallel, delayed
with tqdm_joblib(tqdm(desc="My calculation", total=10)) as progress_bar:
Parallel(n_jobs=16)(delayed(sqrt)(i**2) for i in range(10))
我认为这太棒了，它看起来类似于 tqdm pandas 集成。
【讨论】：
为什么不能简单地使用tqdm？以下对我有用
from joblib import Parallel, delayed
from datetime import datetime
from tqdm import tqdm
def myfun(x):
return x**2
results = Parallel(n_jobs=8)(delayed(myfun)(i) for i in tqdm(range(1000))
100%|██████████| 1000/1000 [00:00<00:00, 10563.37it/s]
【讨论】：
myfun 的开头插入time.sleep(1)，您会发现tqdm 进度几乎立即完成，但results 需要几秒钟才能填充。
您链接到的文档指出Parallel 有一个可选的进度表。它是通过使用multiprocessing.Pool.apply_async提供的callback关键字参数实现的：
# This is inside a dispatch function
self._lock.acquire()
job = self._pool.apply_async(SafeFunction(func), args,
kwargs, callback=CallBack(self.n_dispatched, self))
self._jobs.append(job)
self.n_dispatched += 1
...
class CallBack(object):
""" Callback used by parallel: it is used for progress reporting, and
to add data to be processed
"""
def __init__(self, index, parallel):
self.parallel = parallel
self.index = index
def __call__(self, out):
self.parallel.print_progress(self.index)
if self.parallel._original_iterable:
self.parallel.dispatch_next()
这里是print_progress：
def print_progress(self, index):
elapsed_time = time.time() - self._start_time
# This is heuristic code to print only 'verbose' times a messages
# The challenge is that we may not know the queue length
if self._original_iterable:
if _verbosity_filter(index, self.verbose):
return
self._print('Done %3i jobs | elapsed: %s',
(index + 1,
short_format_time(elapsed_time),
))
else:
# We are finished dispatching
queue_length = self.n_dispatched
# We always display the first loop
if not index == 0:
# Display depending on the number of remaining items
# A message as soon as we finish dispatching, cursor is 0
cursor = (queue_length - index + 1
- self._pre_dispatch_amount)
frequency = (queue_length // self.verbose) + 1
is_last_item = (index + 1 == queue_length)
if (is_last_item or cursor % frequency):
return
remaining_time = (elapsed_time / (index + 1) *
(self.n_dispatched - index - 1.))
self._print('Done %3i out of %3i | elapsed: %s remaining: %s',
(index + 1,
queue_length,
short_format_time(elapsed_time),
short_format_time(remaining_time),
))
老实说，他们实现这一点的方式有点奇怪 - 它似乎假设任务将始终按照它们开始的顺序完成。转到print_progress 的index 变量只是作业实际开始时的self.n_dispatched 变量。因此，启动的第一个作业将始终以 0 的 index 结束，即使说第三个作业首先完成。这也意味着他们实际上并没有跟踪已完成 工作的数量。所以没有实例变量供您监控。
我认为最好的办法是制作自己的 CallBack 类和猴子补丁 Parallel：
from math import sqrt
from collections import defaultdict
from joblib import Parallel, delayed
class CallBack(object):
completed = defaultdict(int)
def __init__(self, index, parallel):
self.index = index
self.parallel = parallel
def __call__(self, index):
CallBack.completed[self.parallel] += 1
print("done with {}".format(CallBack.completed[self.parallel]))
if self.parallel._original_iterable:
self.parallel.dispatch_next()
import joblib.parallel
joblib.parallel.CallBack = CallBack
if __name__ == "__main__":
print(Parallel(n_jobs=2)(delayed(sqrt)(i**2) for i in range(10)))
输出：
done with 1
done with 2
done with 3
done with 4
done with 5
done with 6
done with 7
done with 8
done with 9
done with 10
[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]
这样，每当作业完成时都会调用您的回调，而不是默认的。
【讨论】：
__call__ 时自定义参数吗？ （对整个 Parallel 类进行子分类可能是一种方法，但对我来说很重）。
扩展 dano 对最新版本的 joblib 库的回答。内部实现有一些变化。
from joblib import Parallel, delayed
from collections import defaultdict
# patch joblib progress callback
class BatchCompletionCallBack(object):
completed = defaultdict(int)
def __init__(self, time, index, parallel):
self.index = index
self.parallel = parallel
def __call__(self, index):
BatchCompletionCallBack.completed[self.parallel] += 1
print("done with {}".format(BatchCompletionCallBack.completed[self.parallel]))
if self.parallel._original_iterator is not None:
self.parallel.dispatch_next()
import joblib.parallel
joblib.parallel.BatchCompletionCallBack = BatchCompletionCallBack
【讨论】：
TLDR 解决方案：
使用 python 3.5 与 joblib 0.14.0 和 tqdm 4.46.0 一起使用。感谢 frenzykryger 提供 contextlib 建议，感谢 dano 和 Connor 提供猴子修补想法。
import contextlib
import joblib
from tqdm import tqdm
from joblib import Parallel, delayed
@contextlib.contextmanager
def tqdm_joblib(tqdm_object):
"""Context manager to patch joblib to report into tqdm progress bar given as argument"""
def tqdm_print_progress(self):
if self.n_completed_tasks > tqdm_object.n:
n_completed = self.n_completed_tasks - tqdm_object.n
tqdm_object.update(n=n_completed)
original_print_progress = joblib.parallel.Parallel.print_progress
joblib.parallel.Parallel.print_progress = tqdm_print_progress
try:
yield tqdm_object
finally:
joblib.parallel.Parallel.print_progress = original_print_progress
tqdm_object.close()
你可以像 frenzykryger 描述的那样使用它
import time
def some_method(wait_time):
time.sleep(wait_time)
with tqdm_joblib(tqdm(desc="My method", total=10)) as progress_bar:
Parallel(n_jobs=2)(delayed(some_method)(0.2) for i in range(10))
更长的解释：
Jon 的解决方案实现起来很简单，但它只测量已调度的任务。如果任务耗时较长，则在等待最后一个分派的任务执行完毕时，进度条会一直卡在 100%。
从 dano 和 Connor 改进而来的 frenzykryger 的上下文管理器方法更好，但在任务完成之前也可以使用 ImmediateResult 调用 BatchCompletionCallBack（参见 Intermediate results from joblib）。这将使我们的计数超过 100%。
我们可以只修补Parallel 中的print_progress 函数，而不是猴子修补BatchCompletionCallBack。无论如何，BatchCompletionCallBack 已经调用了这个print_progress。如果设置了详细信息（即Parallel(n_jobs=2, verbose=100)），print_progress 将打印出已完成的任务，尽管不如 tqdm 好。查看代码，print_progress 是一个类方法，所以它已经有 self.n_completed_tasks 记录我们想要的数字。我们所要做的只是将其与 joblib 的当前进度状态进行比较，并仅在有差异时更新。
这是使用 python 3.5 在 joblib 0.14.0 和 tqdm 4.46.0 中测试的。
【讨论】：
文字进度条
对于那些想要文本进度条而不需要像 tqdm 这样的附加模块的人来说，还有一个变体。 Joblib=0.11 的实际值，Linux 上的 python 3.5.2 于 16.04.2018 并在子任务完成时显示进度。
重新定义原生类：
class BatchCompletionCallBack(object):
# Added code - start
global total_n_jobs
# Added code - end
def __init__(self, dispatch_timestamp, batch_size, parallel):
self.dispatch_timestamp = dispatch_timestamp
self.batch_size = batch_size
self.parallel = parallel
def __call__(self, out):
self.parallel.n_completed_tasks += self.batch_size
this_batch_duration = time.time() - self.dispatch_timestamp
self.parallel._backend.batch_completed(self.batch_size,
this_batch_duration)
self.parallel.print_progress()
# Added code - start
progress = self.parallel.n_completed_tasks / total_n_jobs
print(
"\rProgress: [{0:50s}] {1:.1f}%".format('#' * int(progress * 50), progress*100)
, end="", flush=True)
if self.parallel.n_completed_tasks == total_n_jobs:
print('\n')
# Added code - end
if self.parallel._original_iterator is not None:
self.parallel.dispatch_next()
import joblib.parallel
import time
joblib.parallel.BatchCompletionCallBack = BatchCompletionCallBack
使用作业总数定义全局常量：
total_n_jobs = 10
这将导致如下结果：
Progress: [######################################## ] 80.0%
【讨论】：
__call__：``` time_remaining = (this_batch_duration / self.batch_size) * (total_n_jobs - self.parallel.n_completed_tasks) print("\rProgress: [{0 :50s}] {1:.1f}% est {2:1f}mins left".format('#' * int(progress * 50), progress*100, time_remaining/60) , end="", flush=真）```
这是您问题的另一个答案，语法如下：
aprun = ParallelExecutor(n_jobs=5)
a1 = aprun(total=25)(delayed(func)(i ** 2 + j) for i in range(5) for j in range(5))
a2 = aprun(total=16)(delayed(func)(i ** 2 + j) for i in range(4) for j in range(4))
a2 = aprun(bar='txt')(delayed(func)(i ** 2 + j) for i in range(4) for j in range(4))
a2 = aprun(bar=None)(delayed(func)(i ** 2 + j) for i in range(4) for j in range(4))
【讨论】：
在 Jupyter 中，tqdm 每次输出时都会在输出中开始一个新行。 所以对于 Jupyter Notebook，它将是：
用于 Jupyter 笔记本。 不睡觉：
from joblib import Parallel, delayed
from datetime import datetime
from tqdm import notebook
def myfun(x):
return x**2
results = Parallel(n_jobs=8)(delayed(myfun)(i) for i in notebook.tqdm(range(1000)))
100% 1000/1000 [00:06
随着时间.sleep：
from joblib import Parallel, delayed
from datetime import datetime
from tqdm import notebook
from random import randint
import time
def myfun(x):
time.sleep(randint(1, 5))
return x**2
results = Parallel(n_jobs=7)(delayed(myfun)(i) for i in notebook.tqdm(range(100)))
我目前使用的是什么而不是 joblib.Parallel：
import concurrent.futures
from tqdm import notebook
from random import randint
import time
iterable = [i for i in range(50)]
def myfun(x):
time.sleep(randint(1, 5))
return x**2
def run(func, iterable, max_workers=8):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
results = list(notebook.tqdm(executor.map(func, iterable), total=len(iterable)))
return results
run(myfun, iterable)
【讨论】：