Jak zrównoleglić prostą pętlę Pythona?

Jest to prawdopodobnie trywialne pytanie, ale jak zrównoleglić następującą pętlę w pythonie?

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

Wiem, jak uruchomić pojedyncze wątki w Pythonie, ale nie wiem, jak „zbierać” wyniki.

Wiele procesów byłoby również w porządku - cokolwiek jest najłatwiejsze w tym przypadku. Używam obecnie Linuksa, ale kod powinien również działać w systemach Windows i Mac.

Jaki jest najprostszy sposób na zrównoleglenie tego kodu?

Używanie wielu wątków w CPython nie zapewni lepszej wydajności kodu czysto Python ze względu na globalną blokadę interpretera (GIL). Proponuję multiprocessingzamiast tego użyć modułu:

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

Pamiętaj, że to nie zadziała w interaktywnym tłumaczu.

Aby uniknąć zwykłego FUD wokół GIL: W każdym razie użycie wątków w tym przykładzie nie byłoby korzystne. Ty chcesz tutaj, a nie używać nici procesów, ponieważ uniknąć całą masę problemów.

W celu zrównoleglenia prostej pętli for, joblib wnosi wiele wartości do surowego wykorzystania wieloprocesowości. Nie tylko krótka składnia, ale także takie rzeczy, jak przezroczyste wiązanie iteracji, gdy są one bardzo szybkie (aby usunąć narzut) lub przechwytywanie śledzenia procesu potomnego w celu lepszego raportowania błędów.

Oświadczenie: Jestem oryginalnym autorem joblib.

Jaki jest najprostszy sposób na zrównoleglenie tego kodu?

Naprawdę podoba mi się concurrent.futuresto, dostępne w Python3 od wersji 3.2 - i poprzez backport do wersji 2.6 i 2.7 na PyPi .

Możesz używać wątków lub procesów i używać dokładnie tego samego interfejsu.

Wieloprocesowe

Umieść to w pliku - 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()

A oto wynik:

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

Wielowątkowość

Teraz zmienia ProcessPoolExecutorsię ThreadPoolExecutor, i ponownie uruchomić moduł:

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

Teraz wykonałeś zarówno wielowątkowość, jak i wieloprocesowość!

Uwaga na temat wydajności i korzystania z obu razem.

Próbkowanie jest zdecydowanie zbyt małe, aby porównać wyniki.

Podejrzewam jednak, że wielowątkowość będzie szybsza niż wieloprocesowość w ogóle, szczególnie w systemie Windows, ponieważ system Windows nie obsługuje rozwidlania, więc uruchomienie każdego nowego procesu wymaga czasu. W systemie Linux lub Mac będą prawdopodobnie bliżej.

Możesz zagnieździć wiele wątków w wielu procesach, ale zaleca się, aby nie używać wielu wątków do wydzielenia wielu procesów.

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)

Powyższe działa pięknie na moim komputerze (Ubuntu, pakiet joblib został wstępnie zainstalowany, ale można go zainstalować za pośrednictwem pip install joblib).

Zaczerpnięte z https://blog.dominodatalab.com/simple-parallelization/

Korzystanie z Ray ma wiele zalet :

• Możesz pracować równolegle na wielu komputerach oprócz wielu rdzeni (z tym samym kodem).
• Wydajna obsługa danych numerycznych za pośrednictwem pamięci współużytkowanej (i serializacji bez kopiowania).
• Wysoka przepustowość zadań dzięki planowaniu rozproszonemu.
• Odporność na awarie

W twoim przypadku możesz uruchomić Ray i zdefiniować zdalną funkcję

import ray

ray.init()

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

a następnie wywołać to równolegle

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)

Aby uruchomić ten sam przykład w klastrze, jedyną linią, która by się zmieniła, byłoby wywołanie ray.init (). Odpowiednia dokumentacja znajduje się tutaj .

Zauważ, że pomagam rozwijać Ray.

To najłatwiejszy sposób na zrobienie tego!

Możesz użyć asyncio . (Dokumentacja znajduje się tutaj ). Jest wykorzystywany jako podstawa dla wielu asynchronicznych struktur Pythona, które zapewniają wysokowydajne serwery sieciowe i sieciowe, biblioteki połączeń z bazami danych, rozproszone kolejki zadań itp. Ponadto posiada interfejsy API wysokiego i niskiego poziomu, aby poradzić sobie z każdym rodzajem problemu .

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

Teraz ta funkcja będzie uruchamiana równolegle za każdym razem, gdy zostanie wywołana, bez przełączania programu głównego w stan oczekiwania. Możesz go również użyć do równoległego tworzenia pętli. Gdy wywoływana jest pętla for, pętla jest sekwencyjna, ale każda iteracja przebiega równolegle do programu głównego, gdy tylko dojdzie do interpretera. Na przykład:

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

Daje to następujące wyniki:

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

dlaczego nie używasz wątków i jednego muteksu do ochrony jednej globalnej listy?

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

pamiętaj, że będziesz tak szybki jak najwolniejszy wątek

Uważam, że joblibjest ze mną bardzo przydatny. Zobacz następujący przykład:

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: użyj wszystkich dostępnych rdzeni

Powiedzmy, że mamy funkcję asynchroniczną

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

To musi być uruchomione na dużej tablicy. Niektóre atrybuty są przekazywane do programu, a niektóre są używane z właściwości elementu słownika w tablicy.

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

Zerknij na to;

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

To może nie być właściwy sposób, ale zrobiłbym coś takiego;

Aktualny kod;

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

Mam nadzieję, że to pomaga.

Może to być przydatne podczas wdrażania przetwarzania wieloprocesowego i przetwarzania równoległego / rozproszonego w języku Python.

Samouczek YouTube na temat korzystania z pakietu Techila

Techila to pośrednie oprogramowanie do przetwarzania rozproszonego, które integruje się bezpośrednio z Pythonem za pomocą pakietu techila. Funkcja brzoskwini w pakiecie może być użyteczna w równoległych strukturach pętli. (Poniższy fragment kodu pochodzi z forów społeczności 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
    )

dzięki @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'

bardzo prosty przykład przetwarzania równoległego to

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