使用Sklearn搭配Joblib时触发AttributeError: 'NoneType'无'submit'属性

问题重现 #

运行以下代码时会抛出AttributeError: 'NoneType' object has no attribute 'submit'：

import numpy as np
from joblib import Parallel, delayed, parallel_backend
from sklearn.neighbors import NearestNeighbors


def it():    
    df = np.random.randn(1000).reshape(-1,1)
    NearestNeighbors(n_neighbors=2, p=2).fit(df).kneighbors(df)
    yield 1

f = lambda x: 1
with parallel_backend("loky", inner_max_num_threads=1):
    res = Parallel(n_jobs=2)(delayed(f)(p) for p in  it())

已知三种可规避问题的操作：

• 注释掉NearestNeighbors的调用代码
• 将Parallel的n_jobs参数设为1
• 删除parallel_backend上下文管理器

问题原因 #

这是joblib 1.2.0版本中loky后端与scikit-learn内部并行机制的交互bug：当在生成器内部触发scikit-learn的并行操作（如NearestNeighbors.kneighbors内部会启动线程池），同时通过parallel_backend指定loky后端并启用多进程时，线程池状态会被异常重置为None，导致后续调用submit方法失败。

解决方案 #

1. 降级joblib版本
   将joblib降级至1.1.0版本，该版本不存在此交互问题：

   pip install joblib==1.1.0
   

2. 调整并行配置

   • 若无需限制内部线程数，直接移除parallel_backend上下文管理器，使用joblib默认配置
   • 必须使用parallel_backend时，强制scikit-learn内部使用单线程，避免线程池冲突：

     import sklearn
     sklearn.set_config(n_jobs=1)  # 全局设置scikit-learn内部并行数为1
     
     # 后续原代码不变
     import numpy as np
     from joblib import Parallel, delayed, parallel_backend
     from sklearn.neighbors import NearestNeighbors
     
     def it():    
         df = np.random.randn(1000).reshape(-1,1)
         NearestNeighbors(n_neighbors=2, p=2).fit(df).kneighbors(df)
         yield 1
     
     f = lambda x: 1
     with parallel_backend("loky", inner_max_num_threads=1):
         res = Parallel(n_jobs=2)(delayed(f)(p) for p in  it())
     

3. 重构代码结构
   将生成器内部的NearestNeighbors操作移到生成器外部，避免在生成器迭代时触发并行操作：

   import numpy as np
   from joblib import Parallel, delayed, parallel_backend
   from sklearn.neighbors import NearestNeighbors
   
   # 提前执行NearestNeighbors相关操作
   df = np.random.randn(1000).reshape(-1,1)
   NearestNeighbors(n_neighbors=2, p=2).fit(df).kneighbors(df)
   
   def it():    
       yield 1
   
   f = lambda x: 1
   with parallel_backend("loky", inner_max_num_threads=1):
       res = Parallel(n_jobs=2)(delayed(f)(p) for p in it())
   

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