Učinkovito dijeljenje velikih Numpy nizova između procesa u Pythonu

Ovladavanje zajedničkom memorijom za velike prijenose podataka u Pythonu

Rad s velikim skupovima podataka u Pythonu često donosi izazove, osobito kada u igru ​​dođe višeprocesiranje. Masovno dijeljenje numpy nizovi između procesa djeteta i procesa roditelja bez nepotrebnog kopiranja jedna je takva prepreka.

Zamislite da obrađujete znanstvene podatke, financijske modele ili unose strojnog učenja, a svaki skup podataka zauzima značajnu količinu memorije. 🧠 Iako Pythonov višeprocesni modul nudi način stvaranja i upravljanja podređenim procesima, učinkovito dijeljenje podataka poput numpy polja može biti teško.

Ova tema postaje još kritičnija kada razmislite o pisanju ovih velikih skupova podataka u HDF5 datoteku, format poznat po svojoj robusnosti u rukovanju ogromnim količinama strukturiranih podataka. Bez odgovarajućeg upravljanja memorijom riskirate curenje memorije ili pogreške "memorija nije pronađena", što ometa vaš tijek rada.

U ovom ćemo vodiču istražiti koncept zajedničke memorije za numpy nizove, koristeći praktični problem kao naše sidro. S primjerima i savjetima iz stvarnog svijeta naučit ćete kako učinkovito rukovati velikim podacima izbjegavajući uobičajene zamke. Zaronimo! 🚀

Optimiziranje dijeljenja Numpy polja između procesa

Gore navedene skripte usmjerene su na rješavanje izazova dijeljenja velikih sadržaja numpy nizovi između procesa u Pythonu bez dupliciranja podataka. Primarni cilj je učinkovito korištenje zajedničke memorije, osiguravajući učinkovitu komunikaciju i minimalnu upotrebu resursa. Korištenjem Pythona višeprocesiranje i zajedničkih memorijskih modula, rješenje omogućuje podređenim procesima da učitavaju, obrađuju i dijele numpy nizove natrag u nadređeni proces neprimjetno.

U prvoj skripti podređeni proces koristi Zajednička memorija klasa za dodjelu memorije i dijeljenje podataka. Ovaj pristup eliminira potrebu za kopiranjem, što je bitno za rukovanje velikim skupovima podataka. Niz numpy se rekonstruira u zajedničkom memorijskom prostoru, dopuštajući roditeljskom procesu izravan pristup nizu. Korištenje redova čekanja osigurava ispravnu komunikaciju između procesa roditelja i djeteta, kao što je obavještavanje kada se memorija može prekinuti kako bi se izbjeglo curenje.

Alternativna skripta pojednostavljuje upravljanje procesom korištenjem Bazen.karta funkcija koja automatizira stvaranje i spajanje procesa. Svaki podređeni proces učitava svoju odgovarajuću datoteku i koristi zajedničku memoriju za vraćanje pojedinosti niza nadređenom procesu. Ovaj pristup je čišći i lakši za održavanje, posebno kada radite s više datoteka. To je praktično rješenje za zadatke poput znanstvene obrade podataka ili analize slika, gdje se veliki skupovi podataka moraju učinkovito dijeliti.

Razmotrite scenarij iz stvarnog svijeta u kojem istraživački tim obrađuje genomske podatke pohranjene u velikim tekstualnim datotekama. Svaka datoteka sadrži milijune redaka, što dupliciranje čini nepraktičnim zbog ograničenja memorije. Pomoću ovih skripti svaki podređeni proces učitava datoteku, a nadređeni zapisuje podatke u jednu HDF5 datoteku za daljnju analizu. Uz dijeljenu memoriju, tim izbjegava suvišno korištenje memorije, osiguravajući glatkije operacije. 🚀 Ova metoda ne samo da optimizira performanse, već i smanjuje pogreške poput "memorija nije pronađena" ili curenje memorije, što su uobičajene zamke pri rješavanju takvih zadataka. 🧠

from multiprocessing import Process, Queue
from multiprocessing.shared_memory import SharedMemory
import numpy as np
from pathlib import Path
def loadtxt_worker(out_queue, in_queue, filepath):
    dtype = [('chr', 'S10'), ('pos', '<i4'), ('pct', '<f4'), ('c', '<i4'), ('t', '<i4')]
    data = np.loadtxt(filepath, dtype=dtype)
    shm = SharedMemory(create=True, size=data.nbytes)
    shared_array = np.ndarray(data.shape, dtype=dtype, buffer=shm.buf)
    shared_array[:] = data
    out_queue.put({"name": shm.name, "shape": data.shape, "dtype": dtype})
    while True:
        msg = in_queue.get()
        if msg == "done":
            shm.close()
            shm.unlink()
            break
def main():
    filenames = ["data1.txt", "data2.txt"]
    out_queue = Queue()
    in_queue = Queue()
    processes = []
    for file in filenames:
        p = Process(target=loadtxt_worker, args=(out_queue, in_queue, file))
        p.start()
        processes.append(p)
    for _ in filenames:
        msg = out_queue.get()
        shm = SharedMemory(name=msg["name"])
        array = np.ndarray(msg["shape"], dtype=msg["dtype"], buffer=shm.buf)
        print("Array from child:", array)
        in_queue.put("done")
    for p in processes:
        p.join()
if __name__ == "__main__":
    main()

from multiprocessing import Pool, shared_memory
import numpy as np
from pathlib import Path
def load_and_share(file_info):
    filepath, dtype = file_info
    data = np.loadtxt(filepath, dtype=dtype)
    shm = shared_memory.SharedMemory(create=True, size=data.nbytes)
    shared_array = np.ndarray(data.shape, dtype=dtype, buffer=shm.buf)
    shared_array[:] = data
    return {"name": shm.name, "shape": data.shape, "dtype": dtype}
def main():
    dtype = [('chr', 'S10'), ('pos', '<i4'), ('pct', '<f4'), ('c', '<i4'), ('t', '<i4')]
    filenames = ["data1.txt", "data2.txt"]
    file_info = [(file, dtype) for file in filenames]
    with Pool(processes=2) as pool:
        results = pool.map(load_and_share, file_info)
        for res in results:
            shm = shared_memory.SharedMemory(name=res["name"])
            array = np.ndarray(res["shape"], dtype=res["dtype"], buffer=shm.buf)
            print("Shared Array:", array)
            shm.close()
            shm.unlink()
if __name__ == "__main__":
    main()

Poboljšanje dijeljenja podataka u višeprocesnim okruženjima

Jedan kritičan aspekt rada sa veliki numpy nizovi u multiprocesiranju je osiguravanje učinkovite sinkronizacije i upravljanja zajedničkim resursima. Dok je dijeljena memorija moćan alat, potrebno je pažljivo rukovanje njome kako bi se spriječili sukobi i curenje memorije. Pravilan dizajn osigurava da podređeni procesi mogu dijeliti nizove s nadređenim procesom bez nepotrebnog dupliciranja podataka ili pogrešaka.

Drugi ključni čimbenik je dosljedno rukovanje vrstama i oblicima podataka. Kada podređeni proces učitava podatke pomoću numpy.loadtxt, mora se dijeliti u istoj strukturi među procesima. Ovo je posebno važno kada pišete u formate kao što je HDF5, budući da neispravno strukturiranje podataka može dovesti do neočekivanih rezultata ili oštećenih datoteka. Da bi se to postiglo, pohranjivanje metapodataka o nizu—kao što je njegov oblik, dtype i naziv zajedničke memorije—ključno je za besprijekornu rekonstrukciju u nadređenom procesu.

U stvarnim aplikacijama, kao što je obrada velikih klimatskih skupova podataka ili datoteka sekvenciranja genoma, ove tehnike omogućuju istraživačima da rade učinkovitije. Kombiniranjem zajedničke memorije s redovima čekanja za komunikaciju, veliki skupovi podataka mogu se istovremeno obrađivati ​​bez preopterećenja memorije sustava. Na primjer, zamislite obradu satelitskih podataka gdje svaka datoteka predstavlja temperaturu regije tijekom vremena. 🚀 Sustav mora upravljati ovim masivnim nizovima bez uskih grla, osiguravajući glatke i skalabilne performanse za analitičke zadatke. 🌍