多处理:使用tqdm显示进度条

Question

为了使我的代码更"pythonic"和更快,我使用"多处理"和一个map函数发送它a)函数和b)迭代范围.

植入的解决方案(即直接在范围tqdm.tqdm(范围(0,30))上调用tqdm不适用于多处理(如下面的代码所示).

进度条显示从0到100%(当python读取代码？)但它不指示map函数的实际进度.

如何显示一个进度条,指示"地图"功能在哪一步？

from multiprocessing import Pool
import tqdm
import time

def _foo(my_number):
   square = my_number * my_number
   time.sleep(1)
   return square 

if __name__ == '__main__':
   p = Pool(2)
   r = p.map(_foo, tqdm.tqdm(range(0, 30)))
   p.close()
   p.join()

欢迎任何帮助或建议......

Answer 1

使用imap而不是map,它返回已处理值的迭代器.

from multiprocessing import Pool
import tqdm
import time

def _foo(my_number):
   square = my_number * my_number
   time.sleep(1)
   return square 

if __name__ == '__main__':
   with Pool(2) as p:
      r = list(tqdm.tqdm(p.imap(_foo, range(30)), total=30))

Answer 2

抱歉迟到了，但如果您只需要一个并发映射，我将这个功能添加到tqdm>=4.42.0：

from tqdm.contrib.concurrent import process_map  # or thread_map
import time

def _foo(my_number):
   square = my_number * my_number
   time.sleep(1)
   return square 

if __name__ == '__main__':
   r = process_map(_foo, range(0, 30), max_workers=2)

参考资料：https : //tqdm.github.io/docs/contrib.concurrent/和https://github.com/tqdm/tqdm/blob/master/examples/parallel_bars.py

它支持max_workers和chunksize，您也可以轻松地从 切换process_map到thread_map。

Answer 3

找到解决方案:小心!由于多处理,估计时间(每个循环的迭代,总时间等等......可能不稳定)但进度条工作正常.

注意:Pool的上下文管理器仅适用于Python 3.3版

from multiprocessing import Pool
import time
from tqdm import *

def _foo(my_number):
   square = my_number * my_number
   time.sleep(1)
   return square 

if __name__ == '__main__':
    with Pool(processes=2) as p:
        max_ = 30
        with tqdm(total=max_) as pbar:
            for i, _ in tqdm(enumerate(p.imap_unordered(_foo, range(0, max_)))):
                pbar.update()

Answer 4

根据XaviMartínez的答案,我写了这个函数imap_unordered_bar.它的使用方式imap_unordered与显示处理栏的唯一区别相同.

from multiprocessing import Pool
import time
from tqdm import *

def imap_unordered_bar(func, args, n_processes = 2):
    p = Pool(n_processes)
    res_list = []
    with tqdm(total = len(args)) as pbar:
        for i, res in tqdm(enumerate(p.imap_unordered(func, args))):
            pbar.update()
            res_list.append(res)
    pbar.close()
    p.close()
    p.join()
    return res_list

def _foo(my_number):
    square = my_number * my_number
    time.sleep(1)
    return square 

if __name__ == '__main__':
    result = imap_unordered_bar(_foo, range(5))

Answer 5

import multiprocessing as mp
import tqdm


iterable = ... 
num_cpu = mp.cpu_count() - 2 # dont use all cpus.


def func():
    # your logic
    ...


if __name__ == '__main__':
    with mp.Pool(num_cpu) as p:
        list(tqdm.tqdm(p.imap(func, iterable), total=len(iterable)))

Answer 6

对于带有 apply_async 的进度条，我们可以使用以下建议的代码：

https://github.com/tqdm/tqdm/issues/484

import time
import random
from multiprocessing import Pool
from tqdm import tqdm

def myfunc(a):
    time.sleep(random.random())
    return a ** 2

pool = Pool(2)
pbar = tqdm(total=100)

def update(*a):
    pbar.update()

for i in range(pbar.total):
    pool.apply_async(myfunc, args=(i,), callback=update)
pool.close()
pool.join()

Answer 7

您可以p_tqdm改用。

https://github.com/swansonk14/p_tqdm

from p_tqdm import p_map
import time

def _foo(my_number):
   square = my_number * my_number
   time.sleep(1)
   return square 

if __name__ == '__main__':
   r = p_map(_foo, list(range(0, 30)))