您提供的文档指出Parallel
有一个可选的进度条。它是通过使用multiprocessing.Pool.apply_async
提供的callback
关键字参数来实现的:
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
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:
queue_length = self.n_dispatched
if not index == 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
结束。这也意味着他们实际上没有跟踪已完成的工作数量。因此,你无法监视任何实例变量。
我认为你最好自己制作一个回调类,并对 Parallel 进行 monkey patch。
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]
这样,你的回调函数将在作业完成时被调用,而不是使用默认值。
pip install tqdm_joblib
并from tqdm_joblib import tqdm_joblib
即可。 - Labo