如何高效解决超大型DataFrame多进程处理内存不足问题？

我有一个需要执行大量匹配操作的大型DataFrame，此前一直采用下述方法进行处理。但当前我尝试多进程处理的是一个2GB的CSV文件生成的DataFrame，即便仅设置一个分区，我的电脑仍无法正常处理。我推测原因是DataFrame分块进行多进程处理时，所需内存翻倍，导致电脑无法承受。

以下是我目前使用的代码：

def parallelize_dataframe(df, func, additional_param, num_partitions):
    df_split = np.array_split(df, num_partitions)
    results = []
    with ProcessPoolExecutor(max_workers=num_partitions) as executor:
        futures = {executor.submit(func, chunk, additional_param): chunk for chunk in df_split}
        for future in tqdm(futures, total=len(futures), desc="Overall progress"):
            results.append(future.result())
    return pd.concat(results)

1. 分块读取CSV，避免全量加载 #

直接从CSV分块读取处理，不一次性把2GB文件加载到内存，从根源减少内存占用：

def process_csv_chunk(chunk, func, additional_param):
    return func(chunk, additional_param)

def process_large_csv(csv_path, func, additional_param, chunk_size=10_000):
    results = []
    with pd.read_csv(csv_path, chunksize=chunk_size) as reader:
        with ProcessPoolExecutor() as executor:
            futures = [executor.submit(process_csv_chunk, chunk, func, additional_param) for chunk in reader]
            for future in tqdm(futures, total=len(futures), desc="Processing chunks"):
                results.append(future.result())
    return pd.concat(results)

每个进程仅处理一小部分数据，不会出现全量DataFrame被复制到子进程的情况。

2. 用共享内存减少进程间数据复制 #

如果必须先加载全量DataFrame，改用共享内存机制避免数据重复复制：

import pandas as pd
from multiprocessing import shared_memory

def process_shared_chunk(shm_name, shm_shape, shm_dtype, func, additional_param):
    existing_shm = shared_memory.SharedMemory(name=shm_name)
    chunk = pd.DataFrame(np.ndarray(shm_shape, dtype=shm_dtype, buffer=existing_shm.buf))
    result = func(chunk, additional_param)
    existing_shm.close()
    return result

def parallelize_with_shared_memory(df, func, additional_param, num_partitions):
    df_split = np.array_split(df, num_partitions)
    results = []
    with ProcessPoolExecutor(max_workers=num_partitions) as executor:
        futures = []
        for chunk in df_split:
            arr = chunk.to_numpy()
            shm = shared_memory.SharedMemory(create=True, size=arr.nbytes)
            shared_arr = np.ndarray(arr.shape, dtype=arr.dtype, buffer=shm.buf)
            shared_arr[:] = arr[:]
            futures.append(executor.submit(
                process_shared_chunk,
                shm.name, arr.shape, arr.dtype,
                func, additional_param
            ))
            shm.close()
        for future in tqdm(futures, total=len(futures), desc="Overall progress"):
            results.append(future.result())
        # 清理共享内存
        for chunk in df_split:
            shm = shared_memory.SharedMemory(name=shm.name)
            shm.unlink()
    return pd.concat(results)

3. 优化多进程参数 #

• 不要把max_workers设为num_partitions，建议设为CPU核心数的1-2倍，避免进程过多抢占内存。CPU密集型任务设为os.cpu_count()即可，有IO等待的场景可适当提高。

4. 精简匹配操作函数 #

检查你的func函数，砍掉不必要的内存开销：

• 尽量用原地操作（比如df.drop(columns=..., inplace=True)），避免创建冗余中间DataFrame。
• 字符串匹配优先用str.contains这类向量化操作，替代循环逻辑；复杂正则可改用regex库优化。

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