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

95

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.

MidnightLightning
fonte

Respostas:

80

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))
jfs
fonte
7
Eu vejo a impressão somente após a saída do código (não a cada iteração). Você tem uma sugestão?
Hanan Shteingart
@HananShteingart: Funciona bem no meu sistema (Ubuntu) com Python 2 e 3. Usei def do_word(*a): time.sleep(.1)como exemplo. Se não funcionar para você, crie um exemplo de código mínimo completo que demonstre seu problema: descreva usando palavras o que você espera que aconteça e o que acontece, mencione como você executa seu script Python, qual é o seu sistema operacional, versão Python e postar como uma nova pergunta .
jfs
13
Eu tive o mesmo problema que @HananShteingart: é porque eu estava tentando usar Pool.map(). Eu não sabia que única imap() e imap_unordered()trabalho desta forma - a documentação apenas diz "Uma versão mais preguiçosos do mapa ()", mas realmente significa "o subjacente iterador retorna resultados como eles vêm".
simonmacmullen
@simonmacmullen: tanto a pergunta quanto a minha resposta usam imap_unordered(). O problema de Hanan é provavelmente devido a sys.stderr.write('\r..')(sobrescrever a mesma linha para mostrar o progresso).
jfs
2
Também é possível! Eu queria principalmente documentar uma suposição estúpida que fiz - no caso de alguém lendo isso também fazer.
simonmacmullen,
94

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
Tim
fonte
64
e se pool retornar um valor?
Nickpick
11
Criei uma lista vazia chamada resultado antes do loop e, dentro do loop, apenas result.append (x). Tentei fazer isso com 2 processos e usei imap em vez de map e tudo funcionou como eu queria @nickpick
bs7280
2
então minha barra de progresso está iterando para novas linhas em vez de progredir no local, alguma ideia de por que isso pode ser?
Austin
2
não se esqueça depip install tqdm
Sr. T
3
@ bs7280 Por result.append (x) você quis dizer result.append (_)? O que é x?
jason,
27

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.

Reubano
fonte
4
Eu acho que cria um monte de encadeamentos, e cada encadeamento conta de forma independente
nburn42
1
Tenho funções dentro de funções que resultam em erro de decapagem.
ojunk
21

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)
MidnightLightning
fonte
3
Eu testei isso para Python 2.7.6 e rs._number_left parece ser o número de pedaços restantes. Portanto, se rs._chunksize não for 1, rs._number_left não será o número de itens restantes da lista.
Allen de
Onde devo colocar este código? Quer dizer, isso não é executado até que o conteúdo de rsseja conhecido e é um pouco tarde ou não?
Wakan Tanka
@WakanTanka: Vai para o script principal depois de separar os threads extras. No meu exemplo original, ele vai no loop "while", onde rsjá lançou os outros threads.
MidnightLightning de
1
Você poderia editar sua pergunta e / ou resposta para mostrar um exemplo mínimo de trabalho. Não vejo rsem nenhum loop, sou novato em multiprocessamento e isso ajudaria. Muito obrigado.
Wakan Tanka
1
Pelo menos em python 3.5, a solução usando _number_leftnão funciona. _number_leftrepresenta os pedaços que ainda precisam ser processados. Por exemplo, se eu quiser que 50 elementos sejam passados ​​para minha função em paralelo, para um pool de threads com 3 processos, _map_async()crie 10 chunks com 5 elementos cada. _number_leftem seguida, representa quantos desses blocos foram concluídos.
mSSM de
9

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.

Julien Tourille
fonte
tem que mudar: [pool.apply_async(my_function, (x,), callback=results.append) for x in dummy_args]para(pool.apply_async(my_function, (x,), callback=results.append) for x in dummy_args)
David Przybilla
Isso não é verdade. Um objeto gerador não funcionará aqui. Verificado.
swagatam
9

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.

mrapacz
fonte
4

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]
zeawoas
fonte
3

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
Aronstef
fonte
1

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()
Mott The Tuple
fonte