如何解决使用Pool.map()多进程处理大数据时的内存占用过高问题？

嘿，这个内存指数级增长的问题我之前帮不少人踩过坑，核心原因其实是你当前的流程把所有分组数据都攥在内存里，再加上多进程的对象拷贝机制，两者叠加就导致了M(n) ∝ k×n²的内存消耗。咱们一步步来拆解解决：

1. 先砍掉分组阶段的内存累积：别把所有分组DataFrame都存进列表 #

你现在是先分块读文件、分组后把所有结果存成DataFrame列表——这一步就已经把整个14GB数据的分组版本都留在内存里了，哪怕分块读也没用，因为最终所有分组都会被累积。

换个思路：边分块读、边分组、边把分组数据写到临时文件，处理完一个分块就释放它的内存，绝不让所有分组同时待在内存里：

import pandas as pd
from pathlib import Path

# 创建临时目录存分组文件，自动处理目录存在的情况
temp_group_dir = Path("./temp_group_files")
temp_group_dir.mkdir(exist_ok=True)

# 分块读取大文件，这里可以根据你的内存调整chunk_size
chunk_size = 10**4
for chunk in pd.read_csv("your_large_file.csv", chunksize=chunk_size):
    # 按目标列分组
    grouped_chunk = chunk.groupby("your_group_column")
    for group_key, group_df in grouped_chunk:
        # 生成对应分组的临时文件路径（处理特殊字符避免路径错误）
        temp_file = temp_group_dir / f"group_{str(group_key).replace('/', '_')}.csv"
        # 追加写入，第一次写的时候加表头，之后不加
        group_df.to_csv(
            temp_file,
            mode="a",
            header=not temp_file.exists(),
            index=False
        )

这样每个分块处理完就会被GC自动回收，内存里永远只留当前分块的内容，不会累积所有分组。

2. 优化多进程数据传递：别直接传大DataFrame给子进程 #

Pool.map()会把列表里的每个DataFrame序列化后完整拷贝给子进程——主进程里存一份所有分组的列表，每个子进程又要复制一份自己要处理的DataFrame，内存自然就爆炸了。

改成让子进程自己去读临时文件，主进程只传小得可怜的文件路径：

from multiprocessing import Pool
import pandas as pd

def process_single_group(temp_file_path):
    # 子进程自己读取文件，完全不用主进程传大对象
    group_df = pd.read_csv(temp_file_path)
    # 这里写你的具体处理逻辑
    processed_df = group_df.apply(your_processing_function, axis=1)
    # 把处理结果存到输出目录，别返回大对象给主进程
    output_dir = Path("./processed_results")
    output_dir.mkdir(exist_ok=True)
    output_file = output_dir / f"processed_{temp_file_path.name}"
    processed_df.to_csv(output_file, index=False)
    return None

# 收集所有临时分组文件的路径
all_group_files = list(temp_group_dir.glob("group_*.csv"))

# 启动多进程，注意进程数别超过CPU核心数太多
with Pool(processes=4) as pool:
    pool.map(process_single_group, all_group_files)

这样主进程只需要管理一堆字符串路径，内存占用可以忽略不计，子进程按需读取文件，内存压力分散到各个进程，不会出现指数级增长。

3. 进阶方案：用Dask彻底告别手动内存管理 #

如果14GB的数据已经接近你单机内存的极限，不如直接用Dask——它是专为大数据量设计的并行计算框架，完全兼容Pandas语法，还能自动处理分块、并行、内存释放：

import dask.dataframe as dd

# 用Dask读取大文件，自动分块，不用手动指定chunk_size
ddf = dd.read_csv("your_large_file.csv")
# 按目标列分组
grouped_ddf = ddf.groupby("your_group_column")
# 对每个分组应用处理函数，注意要指定meta参数定义返回类型
# 比如你的处理后返回包含col1:str, col2:int的DataFrame，就写meta={"col1": str, "col2": int}
processed_ddf = grouped_ddf.apply(your_processing_function, meta=your_meta_dictionary)
# 计算并保存结果，Dask会自动并行处理，内存占用始终可控
processed_ddf.compute().to_csv("./processed_results/processed_all.csv", index=False)

Dask会把数据拆成多个块，每个块处理完就释放内存，完全不用你手动管临时文件和多进程，内存占用只会和单块数据大小成正比，绝不会出现指数级增长。

4. 几个小细节帮你再省内存 #

• 读取文件时指定dtype，避免Pandas自动推断导致的内存浪费：比如把字符串列设为category类型（如果重复值多），把整数列设为int32而不是默认的int64：

  dtype_config = {"string_col": "category", "int_col": "int32"}
  pd.read_csv("your_large_file.csv", dtype=dtype_config, chunksize=chunk_size)
  

• 处理完记得删除临时文件，别占磁盘空间：

  import shutil
  shutil.rmtree(temp_group_dir)
  

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