多进程共享Pandas DataFrame字典：如何避免内存拷贝占用过高？

这个问题我之前在处理大数据多进程任务时碰到过，本质是Python多进程的**写时复制（Copy-On-Write, COW）**机制在Pandas对象上的“踩坑”：你看到的32GB内存就是16个进程各自复制了一份2GB数据的结果——哪怕你只做只读操作，Pandas内部的一些视图操作也可能隐性触发写操作，导致COW生效。而manager.dict()慢是因为它依赖进程间通信（IPC）同步，每次访问都要序列化/反序列化对象，性能开销极大，完全不适合只读的大数据共享场景。

下面是几种标准的无拷贝共享方案，按实用程度排序：

1. Python 3.8+ 用multiprocessing.shared_memory（最推荐） #

这是标准库原生支持的无拷贝共享方案，直接在进程间共享内存块，不需要序列化数据，性能拉满。

核心逻辑：

• 主进程把每个DataFrame的底层数组存入共享内存
• 记录每个DataFrame的元数据（列名、索引、数据类型、共享内存名称）
• 子进程通过共享内存名称和元数据，直接重建DataFrame（只是创建视图，完全不复制数据）

给你个具体的实现示例：

import multiprocessing as mp
from multiprocessing import shared_memory
import pandas as pd
import numpy as np

def worker(shared_metadata):
    # 子进程：从共享内存重建DataFrame字典
    df_dict = {}
    for key, meta in shared_metadata.items():
        # 连接到主进程创建的共享内存块
        shm = shared_memory.SharedMemory(name=meta['shm_name'])
        # 基于共享内存创建numpy数组（纯视图，无拷贝）
        arr = np.ndarray(meta['shape'], dtype=np.dtype(meta['dtype']), buffer=shm.buf)
        # 用元数据重建DataFrame
        df = pd.DataFrame(arr, columns=meta['columns'], index=meta['index'])
        df_dict[key] = df
        
        # 注意：别在这里关闭共享内存，交给主进程统一处理
        # shm.close()
    
    # 执行你的只读操作
    for key, df in df_dict.items():
        print(f"Worker process: {key} has shape {df.shape}")

if __name__ == '__main__':
    # 你的原始DataFrame字典
    original_df_dict = {
        'user_data': pd.DataFrame(np.random.rand(1_000_000, 10)),
        'product_data': pd.DataFrame(np.random.rand(500_000, 20))
    }
    
    # 准备共享用的元数据和共享内存对象列表
    shared_metadata = {}
    shm_resources = []
    
    for key, df in original_df_dict.items():
        # 取出DataFrame的底层numpy数组
        arr = df.values
        # 创建共享内存块，大小匹配数组
        shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
        # 把数组数据复制到共享内存（主进程仅做这一次复制）
        shm_arr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
        shm_arr[:] = arr[:]
        
        # 保存元数据，供子进程重建用
        shared_metadata[key] = {
            'shm_name': shm.name,
            'shape': arr.shape,
            'dtype': str(arr.dtype),
            'columns': df.columns.tolist(),
            'index': df.index.tolist()
        }
        shm_resources.append(shm)
    
    # 启动16个工作进程
    processes = []
    for _ in range(16):
        p = mp.Process(target=worker, args=(shared_metadata,))
        p.start()
        processes.append(p)
    
    # 等待所有进程完成
    for p in processes:
        p.join()
    
    # 主进程清理共享内存资源
    for shm in shm_resources:
        shm.close()
        shm.unlink()  # 删除共享内存块

2. 内存映射文件（适合超大数据） #

如果你的数据量超大（甚至超过单进程内存），可以用磁盘文件的内存映射来实现共享——进程间通过映射同一个磁盘文件来共享数据，不需要进程间传递任何大对象，完全无拷贝。

核心逻辑：

• 主进程把每个DataFrame的数组和元数据保存到磁盘
• 子进程通过内存映射加载数组，再用元数据重建DataFrame

示例代码：

import multiprocessing as mp
import pandas as pd
import numpy as np
import os

def worker(file_path_map):
    df_dict = {}
    for key, file_path in file_path_map.items():
        # 用numpy内存映射加载数组（无拷贝，直接映射磁盘文件到内存）
        arr = np.load(file_path, mmap_mode='r')
        # 加载之前保存的列名和索引
        columns = np.load(f"{file_path}_cols.npy", allow_pickle=True).tolist()
        index = np.load(f"{file_path}_idx.npy", allow_pickle=True).tolist()
        # 重建DataFrame
        df = pd.DataFrame(arr, columns=columns, index=index)
        df_dict[key] = df
        
        # 执行只读操作
        print(f"Worker process: {key} has shape {df.shape}")

if __name__ == '__main__':
    original_df_dict = {
        'user_data': pd.DataFrame(np.random.rand(1_000_000, 10)),
        'product_data': pd.DataFrame(np.random.rand(500_000, 20))
    }
    file_path_map = {}
    
    # 保存DataFrame的数组和元数据到临时文件
    for key, df in original_df_dict.items():
        arr_path = f".tmp_{key}.npy"
        np.save(arr_path, df.values)
        np.save(f"{arr_path}_cols.npy", df.columns)
        np.save(f"{arr_path}_idx.npy", df.index)
        file_path_map[key] = arr_path
    
    # 启动进程
    processes = [mp.Process(target=worker, args=(file_path_map,)) for _ in range(16)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    
    # 清理临时文件（可选）
    for path in file_path_map.values():
        os.remove(path)
        os.remove(f"{path}_cols.npy")
        os.remove(f"{path}_idx.npy")

3. 用multiprocessing.Array（适合简单一维场景） #

如果你的DataFrame都是同类型的一维数据（或者可以转成一维），可以用multiprocessing.Array创建共享数组，再包装成DataFrame。不过这个方案灵活性差，因为Array只能是一维的，需要手动处理形状转换，只适合简单场景。

关键注意事项 #

• 严格保证只读：哪怕是隐性的写操作（比如直接修改df.values的某个元素）都会触发COW，导致内存暴涨。所有操作都要确保是返回新对象的只读操作（比如df.query()、df.describe()）
• Unix/Linux下的fork模式：如果能100%保证子进程不触发任何写操作，其实fork后默认会共享父进程内存，但Pandas的内部结构太复杂，很容易因为视图操作触发隐性写，所以还是推荐显式共享方案
• 彻底放弃manager.dict()：它是为进程间可修改的共享字典设计的，性能开销极大，完全不适合只读的大数据共享场景

内容的提问来源于stack exchange，提问作者user40780