首页 > 分享 > 使用joblib库，通过并发提升pandas计算效率

使用joblib库，通过并发提升pandas计算效率

萌宠菠菠乐园
2025-04-01 03:32

pandas本身计算是非多进程的，joblib是一种Python的多进程库，可以实现并发和大量磁盘数据的读写。我们把他应用到pandas提升pandas的计算效率。

一、使用场景

由于建立多进程存在开销和各进程间的协助以及通信问题问题，因此满足以下场景使用多进程使用才能加速，否则效率可能低于非多进程方法或者结果错误。

1、数据集较大。通常大于1w行/列，至少大于5000行才有必要使用多进程加速。

2、数据集可以分割，且分割后的各个块计算逻辑不交叉，典型的应用场景是按行分割或者按列分割。按行列分块后，如果各块计算逻辑独立也可以应用多进程。

3、计算逻辑耗时。对于简单的判断、四则运算、内置函数计算等，在数据集不是超大的情况下，使用多进程无法加速，甚至会降低效率。

二、使用方法

多进程流程

joblib库多进程使用方法为

详细版：

list_np = df_dflist(df1,9) tasks = [delayed(函数)(参数) for 参数 in list_np] res=Parallel(n_jobs=3)(tasks) data = pd.concat(res) 1234

简化版：

res = Parallel(n_jobs=3)([delayed(f)(x,y) for x,y in zip(x_list,y_list)] ) data = pd.concat(res) 12

函数版：

list_np = df_dflist(df1,9) def func_pal(list_np): tasks = [delayed(函数)(参数) for 参数 in list_np] works_pal = Parallel(n_jobs=3) res = works_pal(tasks) data = pd.concat(res) return data 12345678

具体步骤如下，其中步骤一和二实现拆分任务功能，步骤三实现多cpu并行分别完成任务功能，步骤四完成结果合并功能。

步骤一、数据集分割，形成可以并发的数据集列表。使用numpy的array_split函数，尽量对原始数据进行均匀切割，否则这种同步阻塞式的多进程库的时间要等到子任务中耗时最长的任务的结束后，所有子任务才会一并返回结果。

函数1、将DataFrame分割后，将各块数据放入列表，列表元素为DataFrame。

def df_dflist(pf_sp,n): Z_array=pf_sp.values ls_np=np.array_split(Z_array,n,axis=0) ls_df=[pd.DataFrame(i) for i in ls_np] return ls_df 12345

函数2、将DataFrame分割后，将各块数据放入列表，列表元素为np。用于后续处理数据为np的场景。

import pandas as pd import numpy as np def df_nplist(pf_sp,n): Z_array=pf_sp.values ls_np=np.array_split(Z_array,n,axis=0) return ls_np 123456

步骤二、建立任务列表。使用delayed函数函数建立任务列表。

tasks = [delayed(func)(i) for i in data_list ] 1

delayed(func)(i) :func为要多进程执行的函数，i为func的参数。delayed(func)(i) 为链式调用，即delayed(func)返回一个可迭代对象（这里的可迭代对象是函数，结果还可以包括其他不可迭代的返回值），i是返回的可迭代对象的参数。delayed(func)(i)通过内部构造返回一个(函数,args,kwargs)这样的元组，但是并不执行函数，留待后续执行，猜测这也是函数命名为delayed的意义。

[delayed(func)(i) for i in data_list ]通过列表推导式，我们可以为列表中的每一个元素(func,args,kwargs)，通过Parallel分配给不同的进程，在每个进程里面执行一个func(args,kwargs)。其中data_list表示存放步骤一种各分块数据集的列表。

步骤三、并发计算。启动并执行并发

通过Parallel函数完成初始并发配置、启动（进程的创建）和执行并发。

启动并发

works_pal = Parallel(n_jobs=3) 1

执行并发

res = works_pal(tasks) 1

启动和执行并发可以合并为：

res=Parallel(n_jobs=3)(tasks) 1

步骤四、合并结果

data = pd.concat(res) 1

三、常用参数

参数的选择决定了并发是否能够执行以及效果，重要的参数有以下几个

3.1、任务数量。任务列表中的任务数量要与cpu数量匹配效果最好，例如我的windows系统i5 4核处理器经过测试，使用joblib一般要设置为3*并发cpu数时效果最好（与默认的pre_dispatch数量一致）。

3.2、并发cpu数量。Parallel函数中n_jobs设置了并发cpu数量，n_jobs为正表示用几个cpu并发，n_jobs=-1表示使用全部cpu，n_jobs=-2表示使用比全部cpu少一个，依次类推。

3.3、并行实现方式。通过backend（backend: str, default: ‘loky’）参数指定并行化后端的实现方法。本机测试时，默认值loky效果最好；取值’threading’时没有启动多进程，为多线程，对于计算密集型任务执行时间与单进程一致或者更低；取值multiprocessing时进程卡死无法执行。

3.4、pre_dispatch和batch_size。均使用默认值，其中pre_dispatch默认值是3n_jobs（经过测试3n_jobs效果最好，这也是任务数量要设置成3*n_jobs的原因），batch_size默认是’auto’。

四、多个参数时使用zip打包函数，输入参数

多个参数都并行

results = Parallel(n_jobs=num_cores)(delayed(function)(i,j) for i,j in zip(a,b)) 1

只有部分参数并行

在链式调用的参数元组中(i,j,常量)写上全部参数，在生成列表推导式的for 后面只加上需要并行的参数。

results = Parallel(n_jobs=num_cores)(delayed(function)(i,j,常量) for i,j in zip(a,b)) 1