¿Cómo paralelizo un bucle simple de Python?

256

Esta es probablemente una pregunta trivial, pero ¿cómo puedo paralelizar el siguiente ciclo en 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)

Sé cómo iniciar subprocesos individuales en Python, pero no sé cómo "recopilar" los resultados.

Múltiples procesos también estarían bien, lo que sea más fácil para este caso. Actualmente estoy usando Linux, pero el código también debería ejecutarse en Windows y Mac.

¿Cuál es la forma más fácil de paralelizar este código?

yo mismo
fuente

Respuestas:

193

El uso de múltiples subprocesos en CPython no le dará un mejor rendimiento para el código de Python puro debido al bloqueo global del intérprete (GIL). Sugiero usar el multiprocessingmódulo en su lugar:

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

Tenga en cuenta que esto no funcionará en el intérprete interactivo.

Para evitar el FUD habitual alrededor del GIL: de todos modos, no habría ninguna ventaja en usar hilos para este ejemplo. Usted desea utilizar procesos de aquí, no temas, porque evitan un montón de problemas.

Sven Marnach
fuente
46
Dado que esta es la respuesta elegida, ¿es posible tener un ejemplo más completo? ¿Cuáles son los argumentos de calc_stuff?
Eduardo Pignatelli
2
@EduardoPignatelli Lea la documentación del multiprocessingmódulo para obtener ejemplos más completos. Pool.map()básicamente funciona como map(), pero en paralelo.
Sven Marnach
3
¿Hay alguna manera de agregar simplemente una barra de carga tqdm a esta estructura de código? He usado tqdm (pool.imap (calc_stuff, range (0, 10 * offset, offset))) pero no obtengo un gráfico de barra de carga completa.
user8188120
@ user8188120 Nunca antes había oído hablar de tqdm, así que lo siento, no puedo evitarlo.
Sven Marnach
Para ver una barra de carga de tqdm, consulte esta pregunta: stackoverflow.com/questions/41920124/…
Johannes
67

Para paralelizar un bucle for simple, joblib aporta mucho valor al uso sin procesar del multiprocesamiento. No solo la sintaxis corta, sino también cosas como el agrupamiento transparente de iteraciones cuando son muy rápidas (para eliminar la sobrecarga) o la captura del rastreo del proceso secundario, para tener un mejor informe de errores.

Descargo de responsabilidad: soy el autor original de joblib.

Gael Varoquaux
fuente
1
Intenté joblib con jupyter, no funciona. Después de la llamada de retraso en paralelo, la página dejó de funcionar.
Jie
1
Hola, tengo un problema al usar joblib ( stackoverflow.com/questions/52166572/… ), ¿tienes alguna idea de cuál puede ser la causa? Muchas gracias.
Ting Sun
¡Parece algo que quiero intentar! ¿Es posible usarlo con un bucle doble, por ejemplo, para i en rango (10): para j en rango (20)
CutePoison
51

¿Cuál es la forma más fácil de paralelizar este código?

Realmente me gusta concurrent.futurespara esto, disponible en Python3 desde la versión 3.2 , y a través de backport a 2.6 y 2.7 en PyPi .

Puede usar hilos o procesos y usar exactamente la misma interfaz.

Multiprocesamiento

Ponga esto en un archivo - 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()

Y aquí está el resultado:

$ 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

Ahora cambie ProcessPoolExecutora ThreadPoolExecutor, y ejecutar el módulo nuevo:

$ 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]

¡Ahora ha realizado múltiples subprocesos y multiprocesamiento!

Nota sobre el rendimiento y el uso de ambos juntos.

El muestreo es demasiado pequeño para comparar los resultados.

Sin embargo, sospecho que el subprocesamiento múltiple será más rápido que el multiprocesamiento en general, especialmente en Windows, ya que Windows no admite la bifurcación, por lo que cada nuevo proceso debe tomarse un tiempo para iniciarse. En Linux o Mac, probablemente estarán más cerca.

Puede anidar múltiples subprocesos dentro de múltiples procesos, pero se recomienda no usar múltiples subprocesos para separar varios procesos.

Aaron Hall
fuente
¿ThreadPoolExecutor omite las limitaciones impuestas por GIL? tampoco necesitaría unirse () para esperar a que los ejecutores terminen o esto se
soluciona
1
No y no, sí a "manejado implícitamente"
Aaron Hall
Por alguna razón, al escalar el problema, el subprocesamiento múltiple es extremadamente rápido, pero el multiprocesamiento genera un montón de procesos atascados (en macOS). ¿Alguna idea de por qué podría ser eso? El proceso contiene solo bucles anidados y matemáticas, nada exótico.
komodovaran_
@komodovaran_ Un proceso es un proceso completo de Python, uno por cada uno, mientras que un subproceso es solo un subproceso de ejecución con su propia pila que comparte el proceso, su código de bytes y todo lo demás que tiene en memoria con todos los otros subprocesos. ¿Eso ayuda? ?
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)

Lo anterior funciona muy bien en mi máquina (Ubuntu, el paquete joblib fue preinstalado, pero se puede instalar a través de pip install joblib).

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

tyrex
fuente
3
Probé su código, pero en mi sistema la versión secuencial de este código tarda aproximadamente medio minuto y la versión paralela anterior tarda 4 minutos. ¿Porque?
shaifali Gupta
3
¡Gracias por tu respuesta! Creo que esta es la forma más elegante de hacer esto en 2019.
Heikki Pulkkinen
2
el multiprocesamiento no es válido para Python 3.x, por lo que esto no funciona para mí.
EngrStudent
2
@EngrStudent No estoy seguro de lo que quiere decir con "no válido". Funciona para Python 3.6.x para mí.
tyrex
@tyrex gracias por compartir! este paquete de trabajo es excelente y el ejemplo me funciona. Sin embargo, en un contexto más complejo tuve un error desafortunadamente. github.com/joblib/joblib/issues/949
Open Food Broker
13

Existen varias ventajas al usar Ray :

  • Puede paralelizar en múltiples máquinas además de múltiples núcleos (con el mismo código).
  • Manejo eficiente de datos numéricos a través de memoria compartida (y serialización de copia cero).
  • Alto rendimiento de tareas con programación distribuida.
  • Tolerancia a fallos.

En su caso, puede iniciar Ray y definir una función remota

import ray

ray.init()

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

y luego invocarlo en 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 ejecutar el mismo ejemplo en un clúster, la única línea que cambiaría sería la llamada a ray.init (). La documentación relevante se puede encontrar aquí .

Tenga en cuenta que estoy ayudando a desarrollar Ray.

Robert Nishihara
fuente
1
Para cualquiera que esté considerando Ray, puede ser relevante saber que no es compatible de forma nativa con Windows. Algunos trucos para que funcione en Windows usando WSL (Windows Subsystem for Linux) son posibles, aunque difícilmente sea listo para usar si quieres usar Windows.
OscarVanL
9

¡Esta es la forma más fácil de hacerlo!

Puedes usar asyncio . (La documentación se puede encontrar aquí ). Se utiliza como base para múltiples marcos asincrónicos de Python que proporcionan servidores de red y web de alto rendimiento, bibliotecas de conexión de bases de datos, colas de tareas distribuidas, etc. Además, tiene API de alto y bajo nivel para resolver cualquier 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

Ahora esta función se ejecutará en paralelo cada vez que se llame sin poner el programa principal en estado de espera. Puede usarlo para paralelizar for loop también. Cuando se solicita un bucle for, aunque el bucle es secuencial, cada iteración se ejecuta en paralelo al programa principal tan pronto como llega el intérprete. Por ejemplo:

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

Esto produce la siguiente salida:

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
Usuario5
fuente
Creo que hay un error tipográfico wrapped()y debería estar en **kwargslugar de*kwargs
jakub-olczyk
¡Uy! Mi error. Corregido!
Usuario5
6

¿Por qué no utiliza hilos y un mutex para proteger una 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

ten en cuenta que serás tan rápido como tu hilo más lento

Jackdoe
fuente
2
Sé que esta es una respuesta muy antigua, por lo que es un fastidio obtener un voto negativo al azar de la nada. Solo voté en contra porque los hilos no paralelan nada. Los subprocesos en Python están vinculados a que solo un subproceso se ejecute en el intérprete a la vez debido al bloqueo global del intérprete, por lo que admiten la programación concurrente, pero no en paralelo como lo solicita OP.
skrrgwasme
3
@skrrgwasme Sé que lo sabes, pero cuando usas las palabras "no van a paralelizar nada", eso puede confundir a los lectores. Si las operaciones toman mucho tiempo porque están vinculadas a IO o duermen mientras esperan un evento, entonces el intérprete se libera para ejecutar los otros hilos, por lo que esto aumentará la velocidad que la gente espera en esos casos. Solo los hilos enlazados a la CPU se ven realmente afectados por lo que dice skrrgwasme.
Jonathan Hartley
5

Encontré que joblibes muy útil conmigo. Por favor vea el siguiente ejemplo:

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: usa todos los núcleos disponibles

miuxu
fuente
14
Ya sabes, es mejor verificar las respuestas ya existentes antes de publicar las tuyas. Esta respuesta también propone su uso joblib.
sanyash
2

Digamos que tenemos una función asíncrona

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

Eso debe ejecutarse en una gran matriz. Algunos atributos se pasan al programa y otros se usan desde la propiedad del elemento del diccionario en la 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
fuente
1

Echa un vistazo a esto;

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

Puede que esta no sea la forma correcta de hacerlo, pero haría algo como;

Código actual;

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 ayude.

MerreM
fuente
1

Esto podría ser útil al implementar multiprocesamiento y computación paralela / distribuida en Python.

Tutorial de YouTube sobre el uso del paquete techila

Techila es un middleware informático distribuido, que se integra directamente con Python utilizando el paquete techila. La función de melocotón en el paquete puede ser útil para paralelizar estructuras de bucle. (El siguiente fragmento de código es de los foros de la comunidad de 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
fuente
1
Si bien este enlace puede responder la pregunta, es mejor incluir aquí las partes esenciales de la respuesta y proporcionar el enlace como referencia. Las respuestas de solo enlace pueden dejar de ser válidas si la página vinculada cambia.
SL Barth - Restablece a Monica el
2
@SLBarth gracias por los comentarios. Agregué un pequeño código de muestra a la respuesta.
TEe
1

gracias @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
fuente
2
-1. Esta es una respuesta de solo código. Sugeriría agregar una explicación que le diga a los lectores qué hace el código que ha publicado, y quizás dónde pueden encontrar información adicional.
starbeamrainbowlabs
-1

ejemplo muy simple de procesamiento paralelo es

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
fuente
3
Aquí no hay paralelismo en el ciclo for, solo está generando un proceso que ejecuta todo el ciclo; esto NO es lo que pretendía el OP.
facuq