使用multiprocessing pool.starmap时CPU占用骤降为0的问题求助

可能的原因 #

• 进程静默异常或死锁：某个DataFrame的处理逻辑抛出未捕获的异常，导致进程挂起；或是处理过程中涉及未正确释放的资源（如锁）引发死锁。
• 内存资源耗尽：40k个DataFrame的批量处理会占用大量内存，当系统内存不足时，进程会被内核强制挂起，表现为CPU占用骤降但进程不终止。
• 多进程与Pandas的隐性冲突：Pandas部分操作依赖全局状态或线程本地存储，在多进程环境下可能出现隐性阻塞。

具体解决步骤 #

1. 添加异常捕获与日志
   在DataFrame处理函数中强制捕获所有异常，记录异常详情及对应DataFrame的索引，快速定位问题来源：

   def process_df(df, df_index):
       try:
           # 你的处理逻辑
           processed_data = ...
           return processed_data
       except Exception as e:
           # 打印并记录异常
           error_msg = f"第{df_index}个DataFrame处理失败: {str(e)}"
           print(error_msg)
           with open("process_error.log", "a", encoding="utf-8") as f:
               f.write(error_msg + "\n")
           return None
   

2. 优化内存占用

   • 调整进程数：如果之前用CPU_COUNT - 2，可以尝试降至CPU_COUNT // 2，降低内存并发负载。
   • 手动释放内存：处理完每个DataFrame后，显式删除对象并触发垃圾回收：

     def process_df(df):
         # 处理逻辑
         result = ...
         # 释放内存
         del df
         import gc
         gc.collect()
         return result
     

   • 实时监控内存：用psutil查看进程内存占用，确认是否存在内存溢出：

     import psutil
     def process_df(df):
         proc = psutil.Process()
         print(f"当前进程内存: {proc.memory_info().rss / 1024 / 1024:.2f} MB")
         # 处理逻辑
         ...
     

3. 调整多进程工具与任务提交方式
   改用concurrent.futures.ProcessPoolExecutor，它的异常处理更透明；同时用imap_unordered替代starmap，能实时获取已完成的任务结果，更早发现卡住的进程：

   from concurrent.futures import ProcessPoolExecutor
   import os
   
   cpu_count = os.cpu_count()
   with ProcessPoolExecutor(max_workers=cpu_count - 2) as executor:
       # 构造包含索引的任务迭代器
       tasks = ((df, idx) for idx, df in enumerate(df_list))
       # 批量提交任务并迭代结果
       for result in executor.map(process_df, *zip(*tasks)):
           # 处理返回结果
           ...
   

4. 排查单个DataFrame的异常
   随机抽取多个DataFrame单独运行处理逻辑，验证是否存在特定DataFrame（如格式异常、数据量过大）导致的阻塞，若有则单独处理这类异常数据。

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