Python3中大型任务并行化报错：无法pickle '_thread.lock'对象

错误原因 #

你在multiprocessing.Pool.starmap中直接传递了df.myfunc——这个方法绑定了DataFrame实例df，而df内部包含无法被序列化（pickle）的线程锁对象。Python多进程需要将任务相关对象序列化后传递给子进程，因此触发该错误。

修复方案 #

方案1：将计算函数改为独立函数（推荐CPU密集型场景） #

把依赖DataFrame实例的myfunc改成独立函数，仅传递计算所需的列数据，避免传递整个DataFrame：

import multiprocessing
import itertools
import numpy as np

# 独立计算函数，实现原df.myfunc的逻辑
def calc_func(col1_data, col2_data):
    # 替换为你的实际计算逻辑，例如计算两列的相似度、统计值等
    result = ...
    return result

def multi_thread_func(df):
    cols = df.schema.names
    length = len(cols)
    a = np.zeros(length * length)
    
    # 提取所有列的本地数据（Pandas用.values，Spark需先collect转本地数据）
    col_data = {col: df[col].values for col in cols}
    
    # 构造任务参数：每个任务对应两组列数据
    tasks = [(col_data[c1], col_data[c2]) for c1, c2 in itertools.product(cols, repeat=2)]
    
    with multiprocessing.Pool() as pool:
        results = pool.starmap(calc_func, tasks)
    
    # 填充结果数组，用np.nan替代None更适配数值数组
    for i, val in enumerate(results):
        a[i] = np.nan if val is None else val
    
    return a

方案2：改用线程池（仅适用于IO密集型场景） #

如果myfunc是IO密集型（比如调用外部API、读写文件），可以用线程池替代进程池，线程共享内存空间，不需要序列化整个DataFrame：

from multiprocessing.dummy import Pool as ThreadPool
import itertools
import numpy as np

def multi_thread_func(df):
    cols = df.schema.names
    length = len(cols)
    a = np.zeros(length * length)

    with ThreadPool() as pool:
        results = pool.starmap(df.myfunc, itertools.product(cols, repeat=2))
    
    for i, val in enumerate(results):
        a[i] = np.nan if val is None else val
    
    return a

方案3：Spark DataFrame优先用原生并行机制 #

如果df是Spark DataFrame，不要用Python多进程——Spark本身是分布式并行框架，直接用Spark UDF、内置函数或分布式计算逻辑实现myfunc，避免将数据拉到本地处理，效率会更高。

注意事项 #

• 多进程场景下，传递给子进程的所有参数必须是可pickle的，禁止传递整个DataFrame实例。
• 数值数组中用np.nan代替None更合理，None会被自动转为0，可能干扰后续计算。

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