在 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 数组的常见问题解答

1. 共享内存对象如何帮助进行多处理？
2. 共享内存允许多个进程访问同一内存块，而无需复制数据，从而提高了大型数据集的效率。
3. 目的是什么 SharedMemory(create=True, size=data.nbytes)？
4. 该命令创建一个专门为 numpy 数组调整大小的共享内存块，从而实现进程之间的数据共享。
5. 我可以避免共享内存中的内存泄漏吗？
6. 是的，通过使用 shm.close() 和 shm.unlink() 一旦不再需要共享内存，就释放并删除它。
7. 为什么是 np.ndarray 与共享内存一起使用？
8. 它允许从共享缓冲区重建 numpy 数组，确保数据可以以其原始结构访问。
9. 不正确管理共享内存会带来哪些风险？
10. 管理不当可能会导致内存泄漏、数据损坏或“内存未找到”等错误。

多处理任务的高效内存共享

对于处理海量数据集的 Python 开发人员来说，在进程之间有效共享大型 numpy 数组是一项关键技能。利用共享内存不仅可以避免不必要的复制，还可以提高性能，尤其是在数据科学或机器学习等内存密集型应用程序中。

通过队列和共享内存等工具，Python 为进程间通信提供了强大的解决方案。无论是处理气候数据还是基因组序列，这些技术都能确保平稳运行，不会出现内存泄漏或数据损坏。通过遵循最佳实践，开发人员可以自信地应对项目中的类似挑战。 🌟