Wie parallelisiere ich eine einfache Python-Schleife?

255

Dies ist wahrscheinlich eine triviale Frage, aber wie parallelisiere ich die folgende Schleife in 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)

Ich weiß, wie man einzelne Threads in Python startet, aber ich weiß nicht, wie man die Ergebnisse "sammelt".

Auch mehrere Prozesse wären in Ordnung - was auch immer für diesen Fall am einfachsten ist. Ich verwende derzeit Linux, aber der Code sollte auch unter Windows und Mac ausgeführt werden.

Was ist der einfachste Weg, um diesen Code zu parallelisieren?

ich mich
quelle

Antworten:

193

Die Verwendung mehrerer Threads unter CPython bietet aufgrund der globalen Interpreter-Sperre (GIL) keine bessere Leistung für reinen Python-Code. Ich schlage vor, stattdessen das multiprocessingModul zu verwenden:

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

Beachten Sie, dass dies im interaktiven Interpreter nicht funktioniert.

Um die übliche FUD um die GIL zu vermeiden: Es wäre ohnehin kein Vorteil, Threads für dieses Beispiel zu verwenden. Sie möchten hier Prozesse verwenden, keine Threads, da diese eine ganze Reihe von Problemen vermeiden.

Sven Marnach
quelle
46
Ist es möglich, ein umfassenderes Beispiel zu haben, da dies die gewählte Antwort ist? Was sind die Argumente von calc_stuff?
Eduardo Pignatelli
2
@EduardoPignatelli Bitte lesen Sie einfach die Dokumentation des multiprocessingModuls für umfassendere Beispiele. Pool.map()funktioniert im Grunde wie map(), aber parallel.
Sven Marnach
3
Gibt es eine Möglichkeit, dieser Codestruktur einfach eine tqdm-Ladeleiste hinzuzufügen? Ich habe tqdm (pool.imap (calc_stuff, range (0, 10 * offset, offset))) verwendet, erhalte jedoch keine vollständige Grafik der Ladeleiste.
user8188120
@ user8188120 Ich habe noch nie von tqdm gehört, also tut mir leid, ich kann dabei nicht helfen.
Sven Marnach
Eine tqdm-Ladeleiste finden Sie in dieser Frage: stackoverflow.com/questions/41920124/…
Johannes
67

Um eine einfache for-Schleife zu parallelisieren, bringt joblib viel Wert auf die rohe Verwendung von Multiprocessing. Nicht nur die kurze Syntax, sondern auch Dinge wie das transparente Bündeln von Iterationen, wenn diese sehr schnell sind (um den Overhead zu beseitigen) oder das Erfassen des Tracebacks des untergeordneten Prozesses, um eine bessere Fehlerberichterstattung zu erzielen.

Haftungsausschluss: Ich bin der ursprüngliche Autor von joblib.

Gael Varoquaux
quelle
1
Ich habe Joblib mit Jupiter versucht, es funktioniert nicht. Nach dem parallel verzögerten Aufruf funktionierte die Seite nicht mehr.
Jie
1
Hallo, ich habe ein Problem mit joblib ( stackoverflow.com/questions/52166572/… ). Haben Sie eine Ahnung, was die Ursache sein könnte? Vielen Dank.
Ting Sun
Scheint etwas zu sein, das ich ausprobieren möchte! Ist es möglich, es mit einer Doppelschleife zu verwenden, z. B. für i im Bereich (10): für j im Bereich (20)
CutePoison
51

Was ist der einfachste Weg, um diesen Code zu parallelisieren?

Ich mag concurrent.futuresdas sehr, verfügbar in Python3 seit Version 3.2 - und über Backport auf 2.6 und 2.7 auf PyPi .

Sie können Threads oder Prozesse verwenden und genau dieselbe Schnittstelle verwenden.

Mehrfachverarbeitung

Legen Sie dies in eine Datei - 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()

Und hier ist die Ausgabe:

$ 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

Wechseln Sie nun ProcessPoolExecutorzu ThreadPoolExecutorund führen Sie das Modul erneut aus:

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

Jetzt haben Sie sowohl Multithreading als auch Multiprocessing durchgeführt!

Hinweis zur Leistung und zur gemeinsamen Verwendung beider.

Die Stichprobe ist viel zu klein, um die Ergebnisse zu vergleichen.

Ich vermute jedoch, dass Multithreading schneller ist als Multiprocessing im Allgemeinen, insbesondere unter Windows, da Windows das Forking nicht unterstützt und jeder neue Prozess einige Zeit zum Starten benötigt. Unter Linux oder Mac werden sie wahrscheinlich näher sein.

Sie können mehrere Threads in mehreren Prozessen verschachteln. Es wird jedoch empfohlen, nicht mehrere Threads zu verwenden, um mehrere Prozesse auszulagern.

Aaron Hall
quelle
Umgeht ThreadPoolExecutor die von GIL auferlegten Einschränkungen? Sie müssten auch nicht beitreten (), um auf den Abschluss der Executoren zu warten, oder wird dies implizit im Kontextmanager
erledigt
1
Nein und nein, ja zu "implizit behandelt"
Aaron Hall
Aus irgendeinem Grund ist Multithreading beim Skalieren des Problems extrem schnell, aber Multiprocessing führt zu einer Reihe von festgefahrenen Prozessen (in macOS). Irgendeine Idee warum das sein könnte? Der Prozess enthält nur verschachtelte Schleifen und Mathematik, nichts Exotisches.
komodovaran_
@komodovaran_ Ein Prozess ist ein vollständiger Python-Prozess, jeweils einer pro Prozess, während ein Thread nur ein Ausführungsthread mit einem eigenen Stapel ist, der den Prozess, seinen Bytecode und alles andere, was er im Speicher hat, mit allen anderen Threads teilt - hilft das? ?
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)

Das obige funktioniert wunderbar auf meinem Computer (Ubuntu, Paket joblib war vorinstalliert, kann aber über installiert werden pip install joblib).

Genommen von https://blog.dominodatalab.com/simple-parallelization/

Tyrex
quelle
3
Ich habe Ihren Code ausprobiert, aber auf meinem System dauert die sequentielle Version dieses Codes ungefähr eine halbe Minute und die obige parallele Version 4 Minuten. Warum so?
Shaifali Gupta
3
Danke für deine Antwort! Ich denke, dies ist der eleganteste Weg, dies im Jahr 2019 zu tun.
Heikki Pulkkinen
2
Multiprocessing ist für Python 3.x nicht gültig, daher funktioniert dies bei mir nicht.
EngrStudent
2
@EngrStudent Nicht sicher, was Sie mit "ungültig" meinen. Es funktioniert für Python 3.6.x für mich.
Tyrex
@tyrex danke fürs teilen! Dieses Joblib-Paket ist großartig und das Beispiel funktioniert für mich. In einem komplexeren Kontext hatte ich leider einen Fehler. github.com/joblib/joblib/issues/949
Open Food Broker
13

Die Verwendung von Ray bietet eine Reihe von Vorteilen :

  • Sie können zusätzlich zu mehreren Kernen (mit demselben Code) mehrere Computer parallelisieren.
  • Effiziente Verarbeitung numerischer Daten über gemeinsam genutzten Speicher (und Serialisierung ohne Kopie).
  • Hoher Aufgabendurchsatz bei verteilter Planung.
  • Fehlertoleranz.

In Ihrem Fall können Sie Ray starten und eine Remote-Funktion definieren

import ray

ray.init()

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

und rufen Sie es dann parallel auf

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)

Um dasselbe Beispiel in einem Cluster auszuführen, würde sich nur der Aufruf von ray.init () ändern. Die entsprechende Dokumentation finden Sie hier .

Beachten Sie, dass ich bei der Entwicklung von Ray helfe.

Robert Nishihara
quelle
1
Für alle, die Ray in Betracht ziehen, kann es relevant sein zu wissen, dass Windows von Haus aus nicht unterstützt wird. Einige Hacks, um es unter Windows mit WSL (Windows Subsystem für Linux) zum Laufen zu bringen, sind möglich, obwohl es kaum sofort einsatzbereit ist, wenn Sie Windows verwenden möchten.
OscarVanL
9

Dies ist der einfachste Weg, dies zu tun!

Sie können Asyncio verwenden . (Dokumentation finden Sie hier ). Es wird als Grundlage für mehrere asynchrone Python-Frameworks verwendet, die leistungsstarke Netzwerk- und Webserver, Datenbankverbindungsbibliotheken, verteilte Aufgabenwarteschlangen usw. bereitstellen. Außerdem verfügt es über APIs auf hoher und niedriger Ebene, um alle Arten von Problemen zu bewältigen .

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

Jetzt wird diese Funktion bei jedem Aufruf parallel ausgeführt, ohne dass das Hauptprogramm in den Wartezustand versetzt wird. Sie können es auch zum Parallelisieren der for-Schleife verwenden. Wenn eine for-Schleife aufgerufen wird, ist die Schleife zwar sequentiell, aber jede Iteration läuft parallel zum Hauptprogramm, sobald der Interpreter dort ankommt. Zum Beispiel:

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

Dies erzeugt folgende Ausgabe:

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
User5
quelle
Ich denke, es gibt einen Tippfehler wrapped()und es sollte **kwargsstatt*kwargs
jakub-olczyk
Hoppla! Mein Fehler. Korrigiert!
User5
6

Warum verwenden Sie keine Threads und einen Mutex, um eine globale Liste zu schützen?

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

Denken Sie daran, Sie werden so schnell sein wie Ihr langsamster Thread

Jackdoe
quelle
2
Ich weiß, dass dies eine sehr alte Antwort ist, daher ist es ein Mist, aus dem Nichts eine zufällige Abwertung zu bekommen. Ich habe nur herabgestimmt, weil Threads nichts parallelisieren. Threads in Python sind aufgrund der globalen Interpretersperre an jeweils nur einen Thread gebunden, der auf dem Interpreter ausgeführt wird. Daher unterstützen sie die gleichzeitige Programmierung, jedoch nicht parallel, wie OP dies anfordert.
skrrgwasme
3
@skrrgwasme Ich weiß, dass Sie das wissen, aber wenn Sie die Worte "sie werden nichts parallelisieren" verwenden, könnte dies die Leser irreführen. Wenn die Vorgänge lange dauern, weil sie an E / A gebunden sind oder schlafen, während sie auf ein Ereignis warten, kann der Interpreter die anderen Threads ausführen. Dies führt zu einer Geschwindigkeitssteigerung, auf die die Benutzer in diesen Fällen hoffen. Nur CPU-gebundene Threads sind wirklich von den Aussagen von skrrgwasme betroffen.
Jonathan Hartley
5

Ich fand joblibes sehr nützlich bei mir. Bitte sehen Sie folgendes Beispiel:

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: Alle verfügbaren Kerne verwenden

miuxu
quelle
14
Sie wissen, es ist besser, bereits vorhandene Antworten zu überprüfen, bevor Sie Ihre eigenen veröffentlichen. Diese Antwort schlägt auch vor, zu verwenden joblib.
Sanyash
2

Angenommen, wir haben eine asynchrone Funktion

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

Das muss auf einem großen Array ausgeführt werden. Einige Attribute werden an das Programm übergeben, andere werden von der Eigenschaft des Wörterbuchelements im Array verwendet.

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
quelle
1

Schau dir das an;

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

Dies ist vielleicht nicht der richtige Weg, aber ich würde so etwas tun.

Tatsächlicher Code;

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

Hoffentlich hilft das.

MerreM
quelle
1

Dies kann nützlich sein, wenn Sie Multiprocessing und Parallel / Distributed Computing in Python implementieren.

YouTube-Tutorial zur Verwendung des Techila-Pakets

Techila ist eine verteilte Computing-Middleware, die mithilfe des Techila-Pakets direkt in Python integriert wird. Die Pfirsichfunktion im Paket kann beim Parallelisieren von Schleifenstrukturen hilfreich sein. (Das folgende Code-Snippet stammt aus den Techila Community-Foren. )

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
quelle
1
Während dieser Link die Frage beantworten kann, ist es besser, die wesentlichen Teile der Antwort hier aufzunehmen und den Link als Referenz bereitzustellen. Nur-Link-Antworten können ungültig werden, wenn sich die verknüpfte Seite ändert.
SL Barth - Wiedereinsetzung Monica
2
@ SLBarth danke für das Feedback. Ich habe der Antwort einen kleinen Beispielcode hinzugefügt.
TEe
1

danke @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
quelle
2
-1. Dies ist eine reine Code-Antwort. Ich würde vorschlagen, eine Erklärung hinzuzufügen, die den Lesern sagt, was der von Ihnen gepostete Code bewirkt und wo sie möglicherweise zusätzliche Informationen finden können.
Starbeamrainbowlabs
-1

sehr einfaches Beispiel für Parallelverarbeitung ist

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
quelle
3
Hier gibt es keine Parallelität in der for-Schleife. Sie erzeugen lediglich einen Prozess, der die gesamte Schleife ausführt. Dies ist NICHT das, was das OP beabsichtigt hat.
Facuq