Python의 프로세스 간에 대규모 Numpy 배열을 효율적으로 공유

복사하지 않고 프로세스 간에 Numpy 배열을 효율적으로 공유

Python 다중 처리 및 공유 메모리를 사용하는 백엔드 솔루션입니다.

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

Python의 다중 처리 풀을 사용한 대체 접근 방식

보다 간편한 관리를 위해 멀티프로세싱 풀을 활용하는 솔루션입니다.

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

Python 다중 처리에서 Numpy 배열 공유에 대한 FAQ

1. 공유 메모리 개체는 다중 처리에 어떻게 도움이 되나요?
2. 공유 메모리를 사용하면 여러 프로세스가 데이터를 복사하지 않고도 동일한 메모리 블록에 액세스할 수 있으므로 대규모 데이터 세트의 효율성이 향상됩니다.
3. 목적은 무엇입니까? SharedMemory(create=True, size=data.nbytes)?
4. 이 명령은 numpy 배열에 맞게 특별히 크기가 지정된 공유 메모리 블록을 생성하여 프로세스 간 데이터 공유를 활성화합니다.
5. 공유 메모리에서 메모리 누수를 방지할 수 있나요?
6. 예, 다음을 사용하여 shm.close() 그리고 shm.unlink() 더 이상 필요하지 않은 공유 메모리를 해제하고 삭제합니다.
7. 왜? np.ndarray 공유 메모리와 함께 사용됩니까?
8. 공유 버퍼에서 numpy 배열을 재구성하여 원래 구조에서 데이터에 액세스할 수 있도록 합니다.
9. 공유 메모리를 제대로 관리하지 않으면 어떤 위험이 있나요?
10. 잘못 관리하면 메모리 누수, 데이터 손상 또는 "메모리를 찾을 수 없음"과 같은 오류가 발생할 수 있습니다.

다중 처리 작업을 위한 효율적인 메모리 공유

프로세스 간에 대규모 numpy 배열을 효율적으로 공유하는 것은 대규모 데이터 세트로 작업하는 Python 개발자에게 중요한 기술입니다. 공유 메모리를 활용하면 불필요한 복사를 피할 수 있을 뿐만 아니라 특히 데이터 과학이나 기계 학습과 같은 메모리 집약적인 애플리케이션에서 성능이 향상됩니다.

대기열 및 공유 메모리와 같은 도구를 통해 Python은 프로세스 간 통신을 위한 강력한 솔루션을 제공합니다. 기후 데이터를 처리하든 게놈 서열을 처리하든 이러한 기술은 메모리 누수나 데이터 손상 없이 원활한 작동을 보장합니다. 모범 사례를 따르면 개발자는 자신의 프로젝트에서 유사한 문제를 자신 있게 해결할 수 있습니다. 🌟