Mostrar o progresso de uma chamada imap_unordered de pool de multiprocessamento Python?

Eu tenho um script que executa com sucesso um conjunto de tarefas de pool de multiprocessamento com uma imap_unordered()chamada:

p = multiprocessing.Pool()
rs = p.imap_unordered(do_work, xrange(num_tasks))
p.close() # No more work
p.join() # Wait for completion

No entanto, meu num_tasksé cerca de 250.000 e, portanto, join()bloqueia o thread principal por 10 segundos ou mais, e eu gostaria de poder ecoar na linha de comando de forma incremental para mostrar que o processo principal não está bloqueado. Algo como:

p = multiprocessing.Pool()
rs = p.imap_unordered(do_work, xrange(num_tasks))
p.close() # No more work
while (True):
  remaining = rs.tasks_remaining() # How many of the map call haven't been done yet?
  if (remaining == 0): break # Jump out of while loop
  print "Waiting for", remaining, "tasks to complete..."
  time.sleep(2)

Existe um método para o objeto de resultado ou o próprio pool que indica o número de tarefas restantes? Tentei usar um multiprocessing.Valueobjeto como contador ( do_workchama uma counter.value += 1ação após realizar sua tarefa), mas o contador atinge apenas ~ 85% do valor total antes de parar o incremento.

Não há necessidade de acessar atributos privados do conjunto de resultados:

from __future__ import division
import sys

for i, _ in enumerate(p.imap_unordered(do_work, xrange(num_tasks)), 1):
    sys.stderr.write('\rdone {0:%}'.format(i/num_tasks))

Meu favorito pessoal - oferece uma pequena barra de progresso e o ETA de conclusão enquanto as coisas são executadas e confirmadas em paralelo.

from multiprocessing import Pool
import tqdm

pool = Pool(processes=8)
for _ in tqdm.tqdm(pool.imap_unordered(do_work, tasks), total=len(tasks)):
    pass

Descobri que o trabalho já estava feito quando tentei verificar o andamento. Isso é o que funcionou para mim usando o tqdm .

pip install tqdm

from multiprocessing import Pool
from tqdm import tqdm

tasks = range(5)
pool = Pool()
pbar = tqdm(total=len(tasks))

def do_work(x):
    # do something with x
    pbar.update(1)

pool.imap_unordered(do_work, tasks)
pool.close()
pool.join()
pbar.close()

Isso deve funcionar com todos os tipos de multiprocessamento, sejam eles bloqueados ou não.

Encontrou uma resposta-me com um pouco mais de cavar: Tomando um olhar para o __dict__do imap_unorderedobjeto de resultado, descobri que tem um _indexatributo que incrementos com cada conclusão da tarefa. Portanto, isso funciona para o registro, envolto no whileloop:

p = multiprocessing.Pool()
rs = p.imap_unordered(do_work, xrange(num_tasks))
p.close() # No more work
while (True):
  completed = rs._index
  if (completed == num_tasks): break
  print "Waiting for", num_tasks-completed, "tasks to complete..."
  time.sleep(2)

No entanto, descobri que trocar o imap_unorderedpor um map_asyncresultou em uma execução muito mais rápida, embora o objeto de resultado seja um pouco diferente. Em vez disso, o objeto de resultado de map_asynctem um _number_leftatributo e um ready()método:

p = multiprocessing.Pool()
rs = p.map_async(do_work, xrange(num_tasks))
p.close() # No more work
while (True):
  if (rs.ready()): break
  remaining = rs._number_left
  print "Waiting for", remaining, "tasks to complete..."
  time.sleep(0.5)

Eu sei que esta é uma questão bastante antiga, mas aqui está o que estou fazendo quando desejo acompanhar a progressão de um conjunto de tarefas em python.

from progressbar import ProgressBar, SimpleProgress
import multiprocessing as mp
from time import sleep

def my_function(letter):
    sleep(2)
    return letter+letter

dummy_args = ["A", "B", "C", "D"]
pool = mp.Pool(processes=2)

results = []

pbar = ProgressBar(widgets=[SimpleProgress()], maxval=len(dummy_args)).start()

r = [pool.apply_async(my_function, (x,), callback=results.append) for x in dummy_args]

while len(results) != len(dummy_args):
    pbar.update(len(results))
    sleep(0.5)
pbar.finish()

print results

Basicamente, você usa apply_async com um callbak (neste caso, é para anexar o valor retornado a uma lista), então você não precisa esperar para fazer outra coisa. Então, dentro de um loop while, você verifica a progressão do trabalho. Nesse caso, adicionei um widget para torná-lo mais bonito.

A saída:

4 of 4                                                                         
['AA', 'BB', 'CC', 'DD']

Espero que ajude.

Conforme sugerido por Tim, você pode usar tqdme imappara resolver esse problema. Acabei de descobrir esse problema e ajustei a imap_unorderedsolução para poder acessar os resultados do mapeamento. Funciona assim:

from multiprocessing import Pool
import tqdm

pool = multiprocessing.Pool(processes=4)
mapped_values = list(tqdm.tqdm(pool.imap_unordered(do_work, range(num_tasks)), total=len(values)))

Caso você não se importe com os valores retornados de seus trabalhos, você não precisa atribuir a lista a nenhuma variável.

para quem procura uma solução simples trabalhando com Pool.apply_async():

from multiprocessing import Pool
from tqdm import tqdm
from time import sleep


def work(x):
    sleep(0.5)
    return x**2

n = 10

p = Pool(4)
pbar = tqdm(total=n)
res = [p.apply_async(work, args=(
    i,), callback=lambda _: pbar.update(1)) for i in range(n)]
results = [p.get() for p in res]

Eu criei uma classe personalizada para criar uma impressão do progresso. Maby, isso ajuda:

from multiprocessing import Pool, cpu_count


class ParallelSim(object):
    def __init__(self, processes=cpu_count()):
        self.pool = Pool(processes=processes)
        self.total_processes = 0
        self.completed_processes = 0
        self.results = []

    def add(self, func, args):
        self.pool.apply_async(func=func, args=args, callback=self.complete)
        self.total_processes += 1

    def complete(self, result):
        self.results.extend(result)
        self.completed_processes += 1
        print('Progress: {:.2f}%'.format((self.completed_processes/self.total_processes)*100))

    def run(self):
        self.pool.close()
        self.pool.join()

    def get_results(self):
        return self.results

Experimente esta abordagem simples baseada em Fila, que também pode ser usada com pool. Esteja ciente de que imprimir qualquer coisa após o início da barra de progresso fará com que ela seja movida, pelo menos para esta barra de progresso específica. (Progresso 1.5 do PyPI)

import time
from progress.bar import Bar

def status_bar( queue_stat, n_groups, n ):

    bar = Bar('progress', max = n)  

    finished = 0
    while finished < n_groups:

        while queue_stat.empty():
            time.sleep(0.01)

        gotten = queue_stat.get()
        if gotten == 'finished':
            finished += 1
        else:
            bar.next()
    bar.finish()


def process_data( queue_data, queue_stat, group):

    for i in group:

        ... do stuff resulting in new_data

        queue_stat.put(1)

    queue_stat.put('finished')  
    queue_data.put(new_data)

def multiprocess():

    new_data = []

    groups = [[1,2,3],[4,5,6],[7,8,9]]
    combined = sum(groups,[])

    queue_data = multiprocessing.Queue()
    queue_stat = multiprocessing.Queue()

    for i, group in enumerate(groups): 

        if i == 0:

            p = multiprocessing.Process(target = status_bar,
                args=(queue_stat,len(groups),len(combined)))
                processes.append(p)
                p.start()

        p = multiprocessing.Process(target = process_data,
        args=(queue_data, queue_stat, group))
        processes.append(p)
        p.start()

    for i in range(len(groups)):
        data = queue_data.get() 
        new_data += data

    for p in processes:
        p.join()