Multitraitement: utilisez tqdm pour afficher une barre de progression

97

Pour rendre mon code plus "pythonique" et plus rapide, j'utilise le "multiprocessing" et une fonction map pour lui envoyer a) la fonction et b) la plage d'itérations.

La solution implantée (c'est-à-dire appeler tqdm directement sur la plage tqdm.tqdm (plage (0, 30)) ne fonctionne pas avec le multitraitement (comme formulé dans le code ci-dessous).

La barre de progression est affichée de 0 à 100% (quand python lit le code?) Mais elle n'indique pas la progression réelle de la fonction map.

Comment afficher une barre de progression qui indique à quelle étape se trouve la fonction «carte»?

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

Toute aide ou suggestion est la bienvenue ...

SciPy
la source
Pouvez-vous publier l'extrait de code de la barre de progression?
Alex
1
Pour les personnes à la recherche d'une solution avec .starmap(): Voici un patch à Poolajouter .istarmap(), qui fonctionnera également avec tqdm.
Darkonaut

Réponses:

127

Utilisez imap au lieu de map, qui renvoie un itérateur de valeurs traitées.

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))
hkyi
la source
13
Une instruction list () englobante attend la fin de l'itérateur. total = est également requis car tqdm ne sait pas combien de temps l'itération durera,
hkyi
13
Existe-t-il une solution similaire pour starmap()?
tarashypka
1
for i in tqdm.tqdm(...): pass peut être plus simple, quelist(tqdm.tqdm)
savfod
1
Cela fonctionne, mais quelqu'un d'autre l'a-t-il fait imprimer en continu la barre de progression sur une nouvelle ligne pour chaque itération?
Dennis Subachev
3
Le comportement est câblé lorsqu'il est spécifique chunk_sizede p.imap. Peut-on tqdmmettre à jour chaque itération au lieu de chaque morceau?
huangbiubiu
49

Solution trouvée: soyez prudent! En raison du multitraitement, le temps d'estimation (itération par boucle, temps total, etc.) peut être instable, mais la barre de progression fonctionne parfaitement.

Remarque: le gestionnaire de contexte pour Pool n'est disponible qu'à partir de la version 3.3 de Python

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 enumerate(p.imap_unordered(_foo, range(0, max_))):
                pbar.update()
SciPy
la source
2
pbar.close()pas nécessaire, il sera fermé automatiquement à la fin dewith
Sagar Kar
5
Le deuxième / tqdmappel intérieur est -il nécessaire ici?
shadowtalker
5
qu'en est-il de la sortie du _foo (mon_nombre) qui est retourné comme "r" en question?
Likak
3
Existe-t-il une solution similaire pour starmap()?
tarashypka
2
@shadowtalker - cela semble fonctionner sans;). Quoi qu'il en soit - imap_unorderedest la clé ici, il donne les meilleures performances et les meilleures estimations de la barre de progression.
Tomasz Gandor
19

Vous pouvez utiliser à la p_tqdmplace.

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)))
Victor Quach
la source
1
Cela fonctionne extrêmement bien et c'était très facile pip install. Cela remplace tqdm pour la plupart de mes besoins
crypdick
Merci Victor;)
Gabriel Romon
p_tqdmest limité à multiprocessing.Pool, non disponible pour les discussions
pateheo
17

Désolé d'être en retard, mais si tout ce dont vous avez besoin est une carte simultanée, la dernière version ( tqdm>=4.42.0) a maintenant ce intégré:

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)

Références: https://tqdm.github.io/docs/contrib.concurrent/ et https://github.com/tqdm/tqdm/blob/master/examples/parallel_bars.py

casper.dcl
la source
Merci pour cela. Fonctionne facilement, bien mieux que toute autre solution que j'ai essayée.
user3340499
Cool (+1), mais jette HBox(children=(FloatProgress(value=0.0, max=30.0), HTML(value='')))Jupyter
Ébe Isaac le
@ Ébe-Isaac voir github.com/tqdm/tqdm/issues/937
casper.dcl le
Je vois un problème avec la discussion pour pirater tqdm_notebook, cependant, je ne peux pas trouver une solution à résoudre pour tqdm.contrib.concurrent.
Ébe Isaac le
8

basé sur la réponse de Xavi Martínez, j'ai écrit la fonction imap_unordered_bar. Il peut être utilisé de la même manière qu'à imap_unorderedla seule différence qu'une barre de traitement est affichée.

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))
Oliver Wilken
la source
3
Cela redessine la barre à chaque étape sur une nouvelle ligne. Comment mettre à jour la même ligne?
misantroop
Solution dans mon cas (Windows / Powershell): Colorama.
misantroop
'pbar.close () non requis, il sera fermé automatiquement à la fin de avec' comme le commentaire que Sagar a fait sur la réponse de @ scipy
Tejas Shetty
0
import multiprocessing as mp
import tqdm


some_iterable = ...

def some_func():
    # your logic
    ...


if __name__ == '__main__':
    with mp.Pool(mp.cpu_count()-2) as p:
        list(tqdm.tqdm(p.imap(some_func, iterable), total=len(iterable)))
dkrynicki
la source
0

Voici mon point de vue lorsque vous avez besoin de récupérer des résultats à partir de vos fonctions d'exécution parallèles. Cette fonction fait plusieurs choses (il y a un autre de mes articles qui l'explique plus loin) mais le point clé est qu'il y a une file d'attente de tâches en attente et une file d'attente de tâches terminées. Lorsque les travailleurs ont terminé avec chaque tâche de la file d'attente en attente, ils ajoutent les résultats dans la file d'attente des tâches terminées. Vous pouvez encapsuler la vérification dans la file d'attente des tâches terminées avec la barre de progression tqdm. Je ne mets pas l'implémentation de la fonction do_work () ici, ce n'est pas pertinent, car le message ici est de surveiller la file d'attente des tâches terminées et de mettre à jour la barre de progression chaque fois qu'un résultat est obtenu.

def par_proc(job_list, num_cpus=None, verbose=False):

# Get the number of cores
if not num_cpus:
    num_cpus = psutil.cpu_count(logical=False)

print('* Parallel processing')
print('* Running on {} cores'.format(num_cpus))

# Set-up the queues for sending and receiving data to/from the workers
tasks_pending = mp.Queue()
tasks_completed = mp.Queue()

# Gather processes and results here
processes = []
results = []

# Count tasks
num_tasks = 0

# Add the tasks to the queue
for job in job_list:
    for task in job['tasks']:
        expanded_job = {}
        num_tasks = num_tasks + 1
        expanded_job.update({'func': pickle.dumps(job['func'])})
        expanded_job.update({'task': task})
        tasks_pending.put(expanded_job)

# Set the number of workers here
num_workers = min(num_cpus, num_tasks)

# We need as many sentinels as there are worker processes so that ALL processes exit when there is no more
# work left to be done.
for c in range(num_workers):
    tasks_pending.put(SENTINEL)

print('* Number of tasks: {}'.format(num_tasks))

# Set-up and start the workers
for c in range(num_workers):
    p = mp.Process(target=do_work, args=(tasks_pending, tasks_completed, verbose))
    p.name = 'worker' + str(c)
    processes.append(p)
    p.start()

# Gather the results
completed_tasks_counter = 0

with tqdm(total=num_tasks) as bar:
    while completed_tasks_counter < num_tasks:
        results.append(tasks_completed.get())
        completed_tasks_counter = completed_tasks_counter + 1
        bar.update(completed_tasks_counter)

for p in processes:
    p.join()

return results
Nick B.
la source
-2

Cette approche est simple et ça marche.

from multiprocessing.pool import ThreadPool
import time
from tqdm import tqdm

def job():
    time.sleep(1)
    pbar.update()

pool = ThreadPool(5)
with tqdm(total=100) as pbar:
    for i in range(100):
        pool.apply_async(job)
    pool.close()
    pool.join()
Vijayabhaskar J
la source