Multiprocessing: Verwenden Sie tqdm, um einen Fortschrittsbalken anzuzeigen

Um meinen Code "pythonischer" und schneller zu machen, verwende ich "Multiprocessing" und eine Kartenfunktion, um ihn a) die Funktion und b) den Bereich der Iterationen zu senden.

Die implantierte Lösung (dh tqdm direkt im Bereich tqdm.tqdm (Bereich (0, 30)) aufrufen) funktioniert nicht mit Multiprocessing (wie im folgenden Code formuliert).

Der Fortschrittsbalken wird von 0 bis 100% angezeigt (wenn Python den Code liest?), Zeigt jedoch nicht den tatsächlichen Fortschritt der Kartenfunktion an.

Wie kann ein Fortschrittsbalken angezeigt werden, der angibt, in welchem ​​Schritt sich die Kartenfunktion befindet?

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

Jede Hilfe oder Anregungen sind willkommen ...

Verwenden Sie imap anstelle von map, wodurch ein Iterator für verarbeitete Werte zurückgegeben wird.

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

Lösung gefunden: Seien Sie vorsichtig! Aufgrund der Mehrfachverarbeitung kann die Schätzzeit (Iteration pro Schleife, Gesamtzeit usw.) instabil sein, aber der Fortschrittsbalken funktioniert einwandfrei.

Hinweis: Der Kontextmanager für Pool ist nur ab Python Version 3.3 verfügbar

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

Sie können p_tqdmstattdessen verwenden.

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

Es tut uns leid, dass Sie zu spät kommen, aber wenn Sie nur eine gleichzeitige Karte benötigen, ist in der neuesten Version ( tqdm>=4.42.0) jetzt Folgendes integriert:

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)

Verweise: https://tqdm.github.io/docs/contrib.concurrent/ und https://github.com/tqdm/tqdm/blob/master/examples/parallel_bars.py

Basierend auf der Antwort von Xavi Martínez habe ich die Funktion geschrieben imap_unordered_bar. Es kann auf die gleiche Weise verwendet werden, imap_unorderedmit dem einzigen Unterschied, dass eine Verarbeitungsleiste angezeigt wird.

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

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

Hier ist meine Einstellung, wann Sie Ergebnisse von Ihren parallel ausgeführten Funktionen zurückerhalten müssen. Diese Funktion erledigt einige Dinge (es gibt einen anderen Beitrag von mir, der dies weiter erklärt), aber der entscheidende Punkt ist, dass eine Warteschlange für anstehende Aufgaben und eine Warteschlange für abgeschlossene Aufgaben vorhanden ist. Wenn die Mitarbeiter mit jeder Aufgabe in der ausstehenden Warteschlange fertig sind, fügen sie die Ergebnisse in die Warteschlange für abgeschlossene Aufgaben ein. Sie können die Prüfung mit der Fortschrittsanzeige tqdm in die Warteschlange für abgeschlossene Aufgaben einschließen. Ich stelle die Implementierung der Funktion do_work () hier nicht ein, sie ist nicht relevant, da hier die Meldung lautet, dass die Warteschlange für abgeschlossene Aufgaben überwacht und der Fortschrittsbalken jedes Mal aktualisiert werden soll, wenn ein Ergebnis vorliegt.

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

Dieser Ansatz ist einfach und funktioniert.

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