Wie parallelisiere ich eine einfache Python-Schleife?

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?

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.

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.

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.

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/

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.

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

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

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

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

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.

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
    )

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'

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