Como paralelo um loop Python simples?

255

Esta é provavelmente uma pergunta trivial, mas como eu paralelo o seguinte loop em python?

# setup output lists
output1 = list()
output2 = list()
output3 = list()

for j in range(0, 10):
    # calc individual parameter value
    parameter = j * offset
    # call the calculation
    out1, out2, out3 = calc_stuff(parameter = parameter)

    # put results into correct output list
    output1.append(out1)
    output2.append(out2)
    output3.append(out3)

Eu sei como iniciar threads únicos em Python, mas não sei como "coletar" os resultados.

Vários processos também seriam bons - o que for mais fácil para este caso. Atualmente, estou usando o Linux, mas o código também deve ser executado no Windows e no Mac.

Qual é a maneira mais fácil de paralelizar esse código?

eu mesmo
fonte

Respostas:

193

O uso de vários encadeamentos no CPython não fornecerá um melhor desempenho para o código Python puro devido ao bloqueio global de intérpretes (GIL). Sugiro usar o multiprocessingmódulo:

pool = multiprocessing.Pool(4)
out1, out2, out3 = zip(*pool.map(calc_stuff, range(0, 10 * offset, offset)))

Observe que isso não funcionará no intérprete interativo.

Para evitar o FUD usual em torno do GIL: Não haveria nenhuma vantagem em usar threads para este exemplo de qualquer maneira. Você deseja usar processos aqui, não threads, porque eles evitam muitos problemas.

Sven Marnach
fonte
46
Como essa é a resposta escolhida, é possível ter um exemplo mais abrangente? Quais são os argumentos de calc_stuff?
Eduardo Pignatelli
2
@EduardoPignatelli Leia a documentação do multiprocessingmódulo para obter exemplos mais abrangentes. Pool.map()basicamente funciona como map(), mas em paralelo.
Sven Marnach 11/04/19
3
Existe uma maneira de simplesmente adicionar uma barra de carregamento tqdm a essa estrutura de código? Eu usei o tqdm (pool.imap (calc_stuff, range (0, 10 * offset, offset))), mas não recebo um gráfico completo da barra de carregamento.
user8188120
@ user8188120 Eu nunca ouvi falar do tqdm antes, desculpe, não posso ajudar com isso.
Sven Marnach 06/07/19
Para uma barra de carregamento do tqdm, consulte esta pergunta: stackoverflow.com/questions/41920124/…
Johannes
67

Para paralelizar um loop for simples, o joblib agrega muito valor ao uso bruto do multiprocessamento. Não apenas a sintaxe curta, mas também coisas como agrupamento transparente de iterações quando elas são muito rápidas (para remover a sobrecarga) ou a captura do retorno do processo filho, para obter melhores relatórios de erros.

Disclaimer: Eu sou o autor original do joblib.

Gael Varoquaux
fonte
1
Eu tentei joblib com jupyter, não está funcionando. Após a chamada com atraso paralelo, a página parou de funcionar.
Jie
1
Olá, estou com um problema ao usar o joblib ( stackoverflow.com/questions/52166572/… ), você tem alguma idéia do que pode ser a causa? Muito obrigado.
Ting Sun
Parece algo que eu quero tentar! É possível usá-lo com um loop duplo, por exemplo, para i no intervalo (10): para j no intervalo (20)
CutePoison
51

Qual é a maneira mais fácil de paralelizar esse código?

Eu realmente gosto concurrent.futuresdisso, disponível no Python3 desde a versão 3.2 - e via backport para 2.6 e 2.7 no PyPi .

Você pode usar threads ou processos e usar exatamente a mesma interface.

Multiprocessamento

Coloque isso em um arquivo - futuretest.py:

import concurrent.futures
import time, random               # add some random sleep time

offset = 2                        # you don't supply these so
def calc_stuff(parameter=None):   # these are examples.
    sleep_time = random.choice([0, 1, 2, 3, 4, 5])
    time.sleep(sleep_time)
    return parameter / 2, sleep_time, parameter * parameter

def procedure(j):                 # just factoring out the
    parameter = j * offset        # procedure
    # call the calculation
    return calc_stuff(parameter=parameter)

def main():
    output1 = list()
    output2 = list()
    output3 = list()
    start = time.time()           # let's see how long this takes

    # we can swap out ProcessPoolExecutor for ThreadPoolExecutor
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for out1, out2, out3 in executor.map(procedure, range(0, 10)):
            # put results into correct output list
            output1.append(out1)
            output2.append(out2)
            output3.append(out3)
    finish = time.time()
    # these kinds of format strings are only available on Python 3.6:
    # time to upgrade!
    print(f'original inputs: {repr(output1)}')
    print(f'total time to execute {sum(output2)} = sum({repr(output2)})')
    print(f'time saved by parallelizing: {sum(output2) - (finish-start)}')
    print(f'returned in order given: {repr(output3)}')

if __name__ == '__main__':
    main()

E aqui está a saída:

$ python3 -m futuretest
original inputs: [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]
total time to execute 33 = sum([0, 3, 3, 4, 3, 5, 1, 5, 5, 4])
time saved by parallellizing: 27.68999981880188
returned in order given: [0, 4, 16, 36, 64, 100, 144, 196, 256, 324]

Multithreading

Agora mude ProcessPoolExecutorpara ThreadPoolExecutore execute o módulo novamente:

$ python3 -m futuretest
original inputs: [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]
total time to execute 19 = sum([0, 2, 3, 5, 2, 0, 0, 3, 3, 1])
time saved by parallellizing: 13.992000102996826
returned in order given: [0, 4, 16, 36, 64, 100, 144, 196, 256, 324]

Agora você fez o multithreading e o multiprocessing!

Nota sobre desempenho e uso dos dois juntos.

A amostragem é muito pequena para comparar os resultados.

No entanto, suspeito que o multithreading será mais rápido que o multiprocessamento em geral, especialmente no Windows, já que o Windows não suporta bifurcação, portanto, cada novo processo precisa levar algum tempo para ser iniciado. No Linux ou Mac, eles provavelmente estarão mais próximos.

Você pode aninhar vários threads em vários processos, mas é recomendável não usar vários threads para desativar vários processos.

Aaron Hall
fonte
ThreadPoolExecutor ignora as limitações impostas pelo GIL? também wouldnt que você precisa para se juntar a (), a fim de esperar pelos executores ao fim ou é este cuidado implicitamente dentro do gerente de contexto
PirateApp
1
Não e não, sim para "manipulado implicitamente"
Aaron Hall
Por alguma razão, ao expandir o problema, o multithreading é extremamente rápido, mas o multiprocessamento gera vários processos bloqueados (no macOS). Alguma idéia do porquê disso? O processo contém apenas loops aninhados e matemática, nada de exótico.
komodovaran_
@komodovaran_ Um processo é um processo Python completo, um por cada, enquanto um thread é apenas um thread de execução com sua própria pilha que compartilha o processo, seu bytecode e tudo o mais que ele tem na memória com todos os outros threads - isso ajuda ?
Aaron Hall
49
from joblib import Parallel, delayed
import multiprocessing

inputs = range(10) 
def processInput(i):
    return i * i

num_cores = multiprocessing.cpu_count()

results = Parallel(n_jobs=num_cores)(delayed(processInput)(i) for i in inputs)
print(results)

O exemplo acima funciona muito bem na minha máquina (Ubuntu, o pacote joblib foi pré-instalado, mas pode ser instalado via pip install joblib).

Retirado de https://blog.dominodatalab.com/simple-parallelization/

tyrex
fonte
3
Eu tentei o seu código, mas no meu sistema a versão seqüencial desse código leva cerca de meio minuto e a versão paralela acima leva 4 minutos. Por quê então?
shaifali Gupta 13/03/19
3
Obrigado pela sua resposta! Eu acho que esta é a maneira mais elegante de fazer isso em 2019.
Heikki Pulkkinen
2
o multiprocessamento não é válido para o Python 3.x, portanto, isso não funciona para mim.
EngrStudent
2
@EngrStudent Não sabe ao certo o que você quer dizer com "inválido". Funciona para o Python 3.6.x para mim.
Tyrex 16/05/19
@tyrex obrigado por compartilhar! este pacote joblib é ótimo e o exemplo funciona para mim. Porém, em um contexto mais complexo, tive um bug, infelizmente. github.com/joblib/joblib/issues/949
Open Food Broker
13

Há várias vantagens em usar o Ray :

  • Você pode paralelizar várias máquinas, além de vários núcleos (com o mesmo código).
  • Manuseio eficiente de dados numéricos por meio de memória compartilhada (e serialização de cópia zero).
  • Alta taxa de transferência de tarefas com agendamento distribuído.
  • Tolerância ao erro.

No seu caso, você pode iniciar o Ray e definir uma função remota

import ray

ray.init()

@ray.remote(num_return_vals=3)
def calc_stuff(parameter=None):
    # Do something.
    return 1, 2, 3

e depois invocá-lo em paralelo

output1, output2, output3 = [], [], []

# Launch the tasks.
for j in range(10):
    id1, id2, id3 = calc_stuff.remote(parameter=j)
    output1.append(id1)
    output2.append(id2)
    output3.append(id3)

# Block until the results have finished and get the results.
output1 = ray.get(output1)
output2 = ray.get(output2)
output3 = ray.get(output3)

Para executar o mesmo exemplo em um cluster, a única linha que mudaria seria a chamada para ray.init (). A documentação relevante pode ser encontrada aqui .

Note que estou ajudando a desenvolver Ray.

Robert Nishihara
fonte
1
Para quem considera o ray, pode ser relevante saber que ele não oferece suporte nativo ao Windows. Alguns hacks para fazê-lo funcionar no Windows usando o WSL (Windows Subsystem para Linux) são possíveis, embora dificilmente seja imediato se você deseja usar o Windows.
OscarVanL
9

Esta é a maneira mais fácil de fazer isso!

Você pode usar asyncio . (A documentação pode ser encontrada aqui ). É usado como base para várias estruturas assíncronas Python que fornecem servidores de rede e servidores de alta performance, bibliotecas de conexão de banco de dados, filas de tarefas distribuídas etc. Além disso, possui APIs de alto e baixo nível para acomodar qualquer tipo de problema .

import asyncio

def background(f):
    def wrapped(*args, **kwargs):
        return asyncio.get_event_loop().run_in_executor(None, f, *args, **kwargs)

    return wrapped

@background
def your_function(argument):
    #code

Agora, essa função será executada em paralelo sempre que chamada, sem colocar o programa principal no estado de espera. Você pode usá-lo para paralelizar o loop for também. Quando chamado para um loop for, o loop é seqüencial, mas toda iteração é executada paralelamente ao programa principal assim que o intérprete chega. Por exemplo:

@background
def your_function(argument):
    time.sleep(5)
    print('function finished for '+str(argument))


for i in range(10):
    your_function(i)


print('loop finished')

Isso produz a seguinte saída:

loop finished
function finished for 4
function finished for 8
function finished for 0
function finished for 3
function finished for 6
function finished for 2
function finished for 5
function finished for 7
function finished for 9
function finished for 1
Usuário5
fonte
Eu acho que há um erro de digitação wrapped()e deve ser em **kwargsvez de*kwargs
jakub-olczyk
Opa! Meu erro. Corrigido!
User5
6

por que você não usa threads e um mutex para proteger uma lista global?

import os
import re
import time
import sys
import thread

from threading import Thread

class thread_it(Thread):
    def __init__ (self,param):
        Thread.__init__(self)
        self.param = param
    def run(self):
        mutex.acquire()
        output.append(calc_stuff(self.param))
        mutex.release()   


threads = []
output = []
mutex = thread.allocate_lock()

for j in range(0, 10):
    current = thread_it(j * offset)
    threads.append(current)
    current.start()

for t in threads:
    t.join()

#here you have output list filled with data

lembre-se, você será tão rápido quanto o seu fio mais lento

jackdoe
fonte
2
Eu sei que esta é uma resposta muito antiga, por isso é uma chatice obter uma votação aleatória do nada. Eu apenas reduzi a votação porque os threads não paralelizam nada. Os encadeamentos no Python são vinculados a apenas um encadeamento em execução no intérprete por vez, devido ao bloqueio global do intérprete; portanto, eles suportam programação simultânea, mas não paralelos, conforme solicitado pelo OP.
Skrrgwasme 03/03
3
@skrrgwasme Eu sei que você sabe disso, mas quando você usa as palavras "elas não paralelizam nada", isso pode enganar os leitores. Se as operações demorarem muito porque estão vinculadas à entrada / saída ou enquanto aguardam um evento, o intérprete é liberado para executar os outros encadeamentos, portanto, isso resultará no aumento de velocidade que as pessoas esperam nesses casos. Somente os encadeamentos vinculados à CPU são realmente afetados pelo que o skrrgwasme diz.
Jonathan Hartley
5

Eu achei joblibmuito útil para mim. Por favor, veja o seguinte exemplo:

from joblib import Parallel, delayed
def yourfunction(k):   
    s=3.14*k*k
    print "Area of a circle with a radius ", k, " is:", s

element_run = Parallel(n_jobs=-1)(delayed(yourfunction)(k) for k in range(1,10))

n_jobs = -1: use todos os núcleos disponíveis

miuxu
fonte
14
Você sabe, é melhor verificar as respostas já existentes antes de postar as suas. Esta resposta também propõe usar joblib.
21719 Sanyash
2

Digamos que temos uma função assíncrona

async def work_async(self, student_name: str, code: str, loop):
"""
Some async function
"""
    # Do some async procesing    

Isso precisa ser executado em uma grande variedade. Alguns atributos estão sendo passados ​​para o programa e outros são usados ​​a partir da propriedade do elemento de dicionário na matriz.

async def process_students(self, student_name: str, loop):
    market = sys.argv[2]
    subjects = [...] #Some large array
    batchsize = 5
    for i in range(0, len(subjects), batchsize):
        batch = subjects[i:i+batchsize]
        await asyncio.gather(*(self.work_async(student_name,
                                           sub['Code'],
                                           loop)
                       for sub in batch))
Amit Teli
fonte
1

Veja isso;

http://docs.python.org/library/queue.html

Pode não ser o caminho certo, mas eu faria algo assim;

Código atual;

from multiprocessing import Process, JoinableQueue as Queue 

class CustomWorker(Process):
    def __init__(self,workQueue, out1,out2,out3):
        Process.__init__(self)
        self.input=workQueue
        self.out1=out1
        self.out2=out2
        self.out3=out3
    def run(self):
            while True:
                try:
                    value = self.input.get()
                    #value modifier
                    temp1,temp2,temp3 = self.calc_stuff(value)
                    self.out1.put(temp1)
                    self.out2.put(temp2)
                    self.out3.put(temp3)
                    self.input.task_done()
                except Queue.Empty:
                    return
                   #Catch things better here
    def calc_stuff(self,param):
        out1 = param * 2
        out2 = param * 4
        out3 = param * 8
        return out1,out2,out3
def Main():
    inputQueue = Queue()
    for i in range(10):
        inputQueue.put(i)
    out1 = Queue()
    out2 = Queue()
    out3 = Queue()
    processes = []
    for x in range(2):
          p = CustomWorker(inputQueue,out1,out2,out3)
          p.daemon = True
          p.start()
          processes.append(p)
    inputQueue.join()
    while(not out1.empty()):
        print out1.get()
        print out2.get()
        print out3.get()
if __name__ == '__main__':
    Main()

Espero que ajude.

MerreM
fonte
1

Isso pode ser útil ao implementar o multiprocessamento e a computação paralela / distribuída no Python.

Tutorial do YouTube sobre o uso do pacote techila

O Techila é um middleware de computação distribuída, que se integra diretamente ao Python usando o pacote techila. A função pêssego na embalagem pode ser útil em estruturas de loop paralelizadas. (O trecho de código a seguir é dos fóruns da comunidade Techila )

techila.peach(funcname = 'theheavyalgorithm', # Function that will be called on the compute nodes/ Workers
    files = 'theheavyalgorithm.py', # Python-file that will be sourced on Workers
    jobs = jobcount # Number of Jobs in the Project
    )
TEe
fonte
1
Embora esse link possa responder à pergunta, é melhor incluir aqui as partes essenciais da resposta e fornecer o link para referência. As respostas somente para links podem se tornar inválidas se a página vinculada for alterada.
SL Barth - Restabelece Monica
2
@SLBarth obrigado pelo feedback. Eu adicionei um pequeno código de amostra à resposta.
tee
1

obrigado @iuryxavier

from multiprocessing import Pool
from multiprocessing import cpu_count


def add_1(x):
    return x + 1

if __name__ == "__main__":
    pool = Pool(cpu_count())
    results = pool.map(add_1, range(10**12))
    pool.close()  # 'TERM'
    pool.join()   # 'KILL'
Felipe de Macêdo
fonte
2
-1. Esta é uma resposta apenas de código. Sugiro adicionar uma explicação que informe aos leitores o que o código que você postou faz e talvez onde eles possam localizar informações adicionais.
starbeamrainbowlabs
-1

Um exemplo muito simples de processamento paralelo é

from multiprocessing import Process

output1 = list()
output2 = list()
output3 = list()

def yourfunction():
    for j in range(0, 10):
        # calc individual parameter value
        parameter = j * offset
        # call the calculation
        out1, out2, out3 = calc_stuff(parameter=parameter)

        # put results into correct output list
        output1.append(out1)
        output2.append(out2)
        output3.append(out3)

if __name__ == '__main__':
    p = Process(target=pa.yourfunction, args=('bob',))
    p.start()
    p.join()
Adil Warsi
fonte
3
Não há paralelismo no loop for aqui, você está apenas gerando um processo que executa o loop inteiro; NÃO é esse o objetivo do OP.
facuq 2/01