Bagaimana cara memparalelkan loop Python sederhana?

Ini mungkin pertanyaan sepele, tetapi bagaimana cara saya memparalelkan loop berikut dalam 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)

Saya tahu cara memulai utas tunggal dengan Python tapi saya tidak tahu cara "mengumpulkan" hasilnya.

Berbagai proses juga akan baik-baik saja - apa pun yang paling mudah untuk kasus ini. Saya menggunakan Linux saat ini tetapi kode harus berjalan di Windows dan Mac juga.

Apa cara termudah untuk memparalelkan kode ini?

Menggunakan banyak utas di CPython tidak akan memberi Anda kinerja yang lebih baik untuk kode Python murni karena kunci juru bahasa global (GIL). Saya sarankan menggunakan multiprocessingmodul sebagai gantinya:

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

Perhatikan bahwa ini tidak akan berfungsi dalam interpreter interaktif.

Untuk menghindari FUD biasa di sekitar GIL: Toh tidak akan ada keuntungan menggunakan utas untuk contoh ini. Anda ingin menggunakan proses di sini, bukan utas, karena mereka menghindari sejumlah masalah.

Untuk memparalelkan simpel untuk loop, joblib membawa banyak nilai untuk penggunaan multiprosesor secara baku. Tidak hanya sintaks pendek, tetapi juga hal-hal seperti pengelompokan iterasi transparan ketika mereka sangat cepat (untuk menghapus overhead) atau menangkap traceback dari proses anak, untuk memiliki pelaporan kesalahan yang lebih baik.

Penafian: Saya adalah penulis asli joblib.

Apa cara termudah untuk memparalelkan kode ini?

Saya sangat suka concurrent.futuresuntuk ini, tersedia dalam Python3 sejak versi 3.2 - dan via backport ke 2.6 dan 2.7 di PyPi .

Anda dapat menggunakan utas atau proses dan menggunakan antarmuka yang sama persis.

Multiprocessing

Masukkan ini ke dalam file - 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()

Dan inilah hasilnya:

$ 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

Sekarang ubah ProcessPoolExecutormenjadi ThreadPoolExecutor, dan jalankan modul lagi:

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

Sekarang Anda telah melakukan multithreading dan multiprocessing!

Catat kinerja dan gunakan keduanya bersama-sama.

Pengambilan sampel terlalu kecil untuk membandingkan hasilnya.

Namun, saya menduga multithreading akan lebih cepat daripada multiprocessing secara umum, terutama pada Windows, karena Windows tidak mendukung forking sehingga setiap proses baru harus mengambil waktu untuk diluncurkan. Di Linux atau Mac mereka mungkin akan lebih dekat.

Anda dapat membuat sarang banyak utas di dalam beberapa proses, tetapi disarankan untuk tidak menggunakan beberapa utas untuk memulai beberapa proses.

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)

Di atas berfungsi dengan baik di mesin saya (Ubuntu, paket joblib sudah diinstal sebelumnya, tetapi dapat diinstal melalui pip install joblib).

Diambil dari https://blog.dominodatalab.com/simple-parallelization/

Ada beberapa keuntungan menggunakan Ray :

• Anda dapat memparalelkan beberapa mesin selain beberapa inti (dengan kode yang sama).
• Penanganan data numerik yang efisien melalui memori bersama (dan serialisasi tanpa salinan).
• Throughput tugas tinggi dengan penjadwalan terdistribusi.
• Toleransi kesalahan.

Dalam kasus Anda, Anda bisa menjalankan Ray dan mendefinisikan fungsi jarak jauh

import ray

ray.init()

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

dan kemudian memohonnya secara paralel

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)

Untuk menjalankan contoh yang sama pada sebuah cluster, satu-satunya baris yang akan berubah adalah panggilan ke ray.init (). Dokumentasi yang relevan dapat ditemukan di sini .

Perhatikan bahwa saya membantu mengembangkan Ray.

Ini adalah cara termudah untuk melakukannya!

Anda dapat menggunakan asyncio . (Dokumentasi dapat ditemukan di sini ). Ini digunakan sebagai dasar untuk beberapa kerangka kerja asynchronous Python yang menyediakan jaringan dan server web berkinerja tinggi, pustaka koneksi basis data, antrian tugas terdistribusi, dll. Ditambah itu memiliki API level tinggi dan level rendah untuk mengakomodasi segala jenis masalah .

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

Sekarang fungsi ini akan dijalankan secara paralel setiap kali dipanggil tanpa meletakkan program utama ke dalam status menunggu. Anda dapat menggunakannya untuk memparalelkan untuk loop juga. Ketika dipanggil untuk loop, meskipun loop berurutan tetapi setiap iterasi berjalan secara paralel dengan program utama segera setelah penerjemah tiba di sana. Misalnya:

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

Ini menghasilkan output sebagai berikut:

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

mengapa Anda tidak menggunakan utas, dan satu mutex untuk melindungi satu daftar 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

perlu diingat, Anda akan secepat thread paling lambat Anda

Saya menemukan joblibsangat berguna dengan saya. Silakan lihat contoh berikut:

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: gunakan semua core yang tersedia

Katakanlah kita memiliki fungsi async

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

Itu perlu dijalankan pada array yang besar. Beberapa atribut sedang diteruskan ke program dan beberapa digunakan dari properti elemen kamus dalam array.

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

Lihatlah ini;

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

Ini mungkin bukan cara yang tepat untuk melakukannya, tetapi saya akan melakukan sesuatu seperti;

Kode aktual;

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

Semoga itu bisa membantu.

Ini bisa berguna ketika menerapkan komputasi multiprosesor dan paralel / terdistribusi dalam Python.

Tutorial YouTube tentang cara menggunakan paket techila

Techila adalah middleware komputasi terdistribusi, yang terintegrasi langsung dengan Python menggunakan paket techila. Fungsi persik dalam paket dapat berguna dalam memparalelkan struktur loop. (Cuplikan kode berikut berasal dari Forum Komunitas 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
    )

terima kasih @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'

contoh yang sangat sederhana dari pemrosesan paralel adalah

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