Windows环境Jupyter中多进程处理大量CSV文件遇无限运行问题求助

问题背景 #

我有一批规格一致、包含数值的CSV文件，对应有日期列表和作者列表。需求是读取文件计算最大值及对应X值，关联日期、作者信息存入DataFrame。因文件约10万份，尝试用多进程提升效率，但在Windows系统的Jupyter Notebook中运行代码时出现无限运行的情况。

原尝试代码：

import needed things

files = [file1, file2,...,fileN]
dates = [date1, ... , dateN]
authors = [author1, ... , authorN]

def func(file):
    #Reads in file, computes func on values
    index = files.index(file)
    d1 = {'Filename': file, 'Max': max, 'X-value': x-values, 'Date': dates[index], 'Author': authors[index]}
    return d1

if __name__ == '__main__':
    with mp.Pool() as pool:
    data = pool.map(FindTriggers, files)
    summaryDF = pd.DataFrame(data)    

核心问题分析 #

1. Windows多进程启动机制冲突：Windows默认用spawn方式创建子进程，会重新导入整个脚本。但Jupyter Notebook的主模块并非__main__，导致if __name__ == '__main__':的防护失效，子进程会重复初始化进程池，陷入无限递归。
2. 低效索引查找：files.index(file)是O(n)复杂度操作，10万级列表中每个进程都执行该操作，会大幅拖慢速度甚至导致进程阻塞。
3. 代码语法/逻辑错误：函数名func和pool.map调用的FindTriggers不匹配；max、x-values是未定义变量，实际运行会触发报错，可能导致进程挂起。

修复后的代码 #

import pandas as pd
import multiprocessing as mp

def process_file(args):
    # 直接接收打包后的参数，避免索引查找
    file, date, author = args
    # 替换为你的实际CSV读取和计算逻辑
    try:
        df = pd.read_csv(file)
        max_val = df['数值列'].max()
        # 取第一个出现最大值的X值
        x_val = df.loc[df['数值列'] == max_val, 'X列'].iloc[0]
        return {
            'Filename': file,
            'Max': max_val,
            'X-value': x_val,
            'Date': date,
            'Author': author
        }
    except Exception as e:
        # 捕获错误，避免单个进程崩溃影响全局
        return {'Filename': file, 'Error': str(e)}

if __name__ == '__main__':
    # 示例列表，替换为你的实际数据
    files = ['file1.csv', 'file2.csv', ..., 'fileN.csv']
    dates = ['2024-01-01', ..., '2024-12-31']
    authors = ['Alice', ..., 'Bob']
    
    # 打包文件、日期、作者为元组列表
    task_args = list(zip(files, dates, authors))
    
    # 指定spawn启动方式，适配Windows和Jupyter环境
    with mp.get_context('spawn').Pool(processes=mp.cpu_count()-1) as pool:
        # 用imap_unordered可更快返回结果（无需保持顺序）
        results = list(pool.imap_unordered(process_file, task_args))
    
    summaryDF = pd.DataFrame(results)

额外优化建议 #

• 分批次处理：10万文件可分成若干批次（如每批1000个），避免一次性加载过多任务导致内存过载。
• 优先使用imap_unordered：如果不需要结果与文件列表顺序一致，imap_unordered会在子进程完成后立即返回结果，比map更高效。
• 避免全局变量依赖：所有需要的参数都通过函数传入，减少子进程与主进程的耦合，避免变量共享问题。
• 添加进度监控：可配合tqdm库显示处理进度，方便跟踪任务状态：

  from tqdm import tqdm
  results = list(tqdm(pool.imap_unordered(process_file, task_args), total=len(task_args)))
  

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