Python内存受限下高效处理数千个CSV文件的最优方案

针对你需要处理数千个CSV文件、内存受限且追求速度的场景，最优方案是多进程并行处理+轻量级文件解析的组合，核心思路是最大化CPU利用率的同时严格控制内存占用，具体实现如下：

核心优化方向 #

• 用多进程突破GIL限制，充分利用多核CPU处理计算密集型的统计任务
• 替换Pandas为Python内置csv模块，避免额外的内存开销和解析耗时
• 采用边处理边写入的策略，避免主进程缓存大量结果占用内存

具体实现代码 #

1. 单个CSV文件的轻量级处理函数 #

这个函数仅用内置模块逐行读取文件，实时计算所需统计值，内存占用仅为单一行数据的大小：

import csv
import os

def process_single_csv(file_path):
    try:
        # 从文件名提取唯一ID（如ABC.csv -> abc）
        file_id = os.path.splitext(os.path.basename(file_path))[0].lower()
        col1_sum = 0.0
        col1_count = 0
        col2_max = -float('inf')
        
        with open(file_path, 'r', newline='', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                # 转换数据类型并更新统计值
                col1_val = float(row['col1'])
                col2_val = int(row['col2'])
                
                col1_sum += col1_val
                col1_count += 1
                
                if col2_val > col2_max:
                    col2_max = col2_val
        
        col1_avg = round(col1_sum / col1_count, 1) if col1_count > 0 else None
        return (file_id, col1_avg, col2_max)
    except Exception as e:
        print(f"处理文件 {file_path} 失败: {str(e)}")
        return None

2. 多进程并行处理+边处理边写入 #

用进程池并行处理所有文件，同时主进程实时写入结果，无需等待所有文件处理完成：

from multiprocessing import Pool

def main(csv_dir, output_file):
    # 获取目标目录下所有CSV文件路径
    csv_files = [
        os.path.join(csv_dir, filename)
        for filename in os.listdir(csv_dir)
        if filename.lower().endswith('.csv')
    ]
    
    # 初始化进程池，进程数设为CPU核心数（避免上下文切换浪费资源）
    with Pool(processes=os.cpu_count()) as pool:
        # 打开输出文件，边处理边写入
        with open(output_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            # 写入表头
            writer.writerow(['id', 'col1_avg', 'col2_max'])
            # 迭代处理结果，过滤异常返回值
            for result in pool.imap_unordered(process_single_csv, csv_files):
                if result is not None:
                    writer.writerow(result)

if __name__ == '__main__':
    # 替换为你的CSV目录和输出文件路径
    main('./csv_files', './final_output.csv')

额外优化建议 #

• 如果文件数量极大，可改用pool.imap（有序）或pool.imap_unordered（无序）的分批处理模式，进一步降低主进程内存占用
• 给进程池设置合理的进程数：一般等于CPU核心数，若磁盘IO较慢，可适当减少进程数避免IO竞争
• 可根据实际情况调整数据类型转换和精度控制（如示例中的round函数）

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