Python并行库joblib之delayed函数与Parallel函数详解

Joblib

Joblib就是一个可以简单地将Python代码转换为并行计算模式的软件包，它可非常简单并行我们的程序，从而提高计算速度。

主要提供了以下功能

• 程序并行
• 用于在每次使用相同的输入参数调用函数时将其返回值缓存
• 数据存储（包括不可哈希的数据和大规模numpy数组）

程序并行

joblib提供了一个简单地程序并行方案，主要有Parallel函数实现，并涉及了一个技巧性的函数delayed。

delayed函数

  以下为delayed函数的源码

def delayed(function):
    """Decorator used to capture the arguments of a function."""
    def delayed_function(*args, **kwargs):
        return function, args, kwargs
    try:
        delayed_function = functools.wraps(function)(delayed_function)
    except AttributeError:
        " functools.wraps fails on some callable objects "
    return delayed_function

*functools.wraps 旨在消除装饰器对原函数造成的影响，即对原函数的相关属性进行拷贝，已达到装饰器不修改原函数的目的。

从功能上来说，可以认为被wrap修饰后的函数与原函数功能完全相同，暂时忽略不计

delayed函数顾名思义就是延迟函数的执行。

根据源码来看，delayed函数保留被修饰的函数function和参数*args, **kwargs，在碰到调用时，并不直接执行函数function(*args, **kwargs),而是返回返回元组(function,args,kwargs)。

返回的这个结果留待其他函数执行，在joblib里具体是与Parallel配合的。

下面我们通过具体例子看一下delayed函数如何工作的

import functools
def delayed(function):
    """Decorator used to capture the arguments of a function."""
    def delayed_function(*args, **kwargs):
        return function, args, kwargs
    try:
        delayed_function = functools.wraps(function)(delayed_function)
    except AttributeError:
        " functools.wraps fails on some callable objects "
    return delayed_function
def f(x,y):
    return x+y
res = delayed(f)(1,y=3)
print(res)

执行结果为：

(<function f at 0x00000000081C29D8>, (1,), {'y': 3})

返回了原始的函数f和调用它是的两个参数。

上面也说过delayed函数其实是一个修饰器，因此上面的代码与下面的写法等价

@delayed    
def f(x,y):  
    return x+y
res = f(1,y=3)
print(res)

*delayed之后并未得到函数的执行结果，我们如果想得到预期的执行结果应该怎么做呢?其实delayed函数主要是与其他函数配合的，我们可以再写一个程序进行计算：

def f(x,y):    
    return x+y
res = delayed(f)(1,y=2)
print(res)
#out: (<function f at 0x00000000081C2510>, (1,), {'y': 2})
foo,args,kwargs = res
final_res = foo(*args,**kwargs)
print(final_res)
#out: 3

我们先解包了delayed之后的结果res，得到了函数f，和参数args,kwargs，然后调用foo(*args,**kwargs)得到最终得到期待的计算结果的final_res。

开始可能很难理解我们为什么要延迟一个函数的调用呢，调用函数不就是为了得到执行结果吗，这样延迟之后不就无法达到我们的预期了？

先别急，看看下面这个例子应该就明白了

假设我们需要对x_list = [1,2,3], y_list = [ -1,-2,-3] 这两个列表进行逐个元素相加。

可能会觉得这不是很容易实现吗，依次遍历两个集合的元素，在循环过程中调用函数f进行元素相加就好了，可能写出的代码如下：

def f(x,y):    
    return x+y
x_list = [1,2,3]
y_list = [-1,-2,-3]
res = []
for x,y in zip(x_list,y_list):
    res.append(f(x,y))
print(res)

这样线性的多次调用函数f当然没问题，但是如果我们并行的执行f的三次调用，也就是同时执行f(1,-1),f(2,-2),f(3,-3)时代码该怎么写呢？我们希望使用三个线程/进程，每个进程分别执行一个函数调用。这时就可以看出delayed函数的用法了。

@delayed    
def f(x,y):    
    return x+y
x_list = [1,2,3]
y_list = [-1,-2,-3]
res = []
for x,y in zip(x_list,y_list):
    res.append(f(x,y))
print(res)

执行之后的结果为：

[(<function f at 0x00000000081C2950>, (1, -1), {}), 
(<function f at 0x00000000081C2950>, (2, -2), {}), 
(<function f at 0x00000000081C2950>, (3, -3), {})]

我们获得了由(函数,args,kwargs)这样的元组组成的列表，我们可以为列表中的每一个元组(f,args,kwargs)分配给不同的线程，在每个线程里面执行一个f(args,kwargs)。这样是不是就可以完成并行的目的了呢?如何将元组分配给不同的线程由函数Parallel实现，后面再讲。我们现在只需要知道的delayed函数只是为了生成(函数,args,kwargs)这样的元组，暂缓函数的执行，方便将各个计算过程分配给不同的线程。

tips:一般我们只想得到一个这样的列表，并不想改变原始函数f，因此上面的修饰器写法一般等价的写为:

 
def f(x,y):    
    return x+y
x_list = [1,2,3]
y_list = [-1,-2,-3]
res = [delayed(f)(x,y) for x,y in zip(x_list,y_list)]
print(res)

Parallel函数

看名字就知道Parallel主要的功能是实现程序并行。

因此在讲Parallel之前，先看一下并行的实际处理流程：

Parallel实际上就是封装了这个任务拆分，并行和结果合并的这个过程。其主要的功能是线程和进程的创建，和多任务执行这个流程。

由于其源码过于复杂，这里只看和调用相关的两个部分:

(1)、初始化部分

Parallel的初始化主要是与程序并行的配置相关，其函数定义为

class joblib.parallel(n_jobs=None, backend=None, verbose=0, timeout=None, pre_dispatch='2 * n_jobs', 
                   batch_size='auto',temp_folder=None, max_nbytes='1M', mmap_mode='r', prefer=None, require=None)

参数解释：（参考：官方文档）

• n_jobs: int, default: None —— 设置并行执行任务的最大数量

当backend="multiprocessing"时指python工作进程的数量，或者backend="threading"时指线程池大小。
 当n_jobs=-1时，使用所有的CPU执行并行计算；
当n_jobs=1时，就不会使用并行代码，即等同于顺序执行，可以在debug情况下使用；
当n_jobs<-1时，将会使用(n_cpus + 1 + n_jobs)个CPU，例如n_jobs=-2时，将会使用n_cpus-1个CPU核，其中n_cpus为CPU核的数量；
当n_jobs=None的情况等同于n_jobs=1。

• backend: str,  default: 'loky' —— 指定并行化后端的实现方法

backend='loky': 在与Python进程交换输入和输出数据时，可导致一些通信和内存开销。
 
backend='multiprocessing': 基于multiprocessing.Pool的后端，鲁棒性不如loky。
 
backend='threading': threading是一个开销非常低的backend。但是如果被调用的函数大量依赖于Python对象，它就会受到Python全局解释器(GIL)锁的影响。当执行瓶颈是显式释放GIL的已编译扩展时，
“threading”非常有用(例如，封装在“with nogil”块中的Cython循环，或者对库(如NumPy)的大量调用)。

• verbose: int, 可选项 —— 执行期间的信息显示（ps:这个参数实际上有点类似log的日志级别，数字越大，显示的程序执行过程越详细）

信息级别:如果非零，则打印进度消息。超过50，输出被发送到stdout。消息的频率随着信息级别的增加而增加。如果大于10，则报告所有迭代。

• timeout: float, 可选项 —— 任务运行时间限制

timeout仅用在n_jobs != 1的情况下，用来限制每个任务完成的时间，如果任何任务的执行超过这个限制值，将会引发“TimeOutError”错误。

• pre_dispatch: {'all', integer, or expression, as in '3*n_jobs'}

预先分派的(任务的)批数(batches)。默认设置是“2 * n_jobs”。

ps: 这个参数有点难理解，直观来说一下吧。pre_dispatch是预派遣的意思，就是提前先把任务派遣给各个处理器。

注意一下，这里的派遣并不是口头的安排任务，而是把任务和任务对应的数据(划重点)也发送给处理器。假设我们总共有12，3个处理器，如果不设置pre_dispatch，那么程序在开始时会一次性将所有任务都分配出去，一个处理器领到了四个任务。但是这个派遣过程需要为每个任务准备相应的数据，需要处理时间和占据内存空间。 

一次分配任务过多可能造成的后果就是准备数据的时间变长和可能造成内存爆炸。所以设置一下预派遣任务的数量，减小预处理时间和内存占用，保证处理器不空转就行。

• batch_size: int or 'auto', default: 'auto' —— 一次性分派给每个worker的原子任务的数量

当单个原子任务执行非常快时，由于开销的原因，使用dispatching的worker可能比顺序计算慢。一起进行批量快速计算可以缓解这种情况。
 
“auto”策略会跟踪一个批处理完成所需的时间，并动态调整batch_size大小，使用启发式方法将时间保持在半秒以内。初始batch_size为1。
 
batch_size="auto"且backend="threading时，将一次分派一个任务的batches，因为threading后端有非常小的开销，使用更大的batch_size在这种情况下没有证明带来任何好处。

这个可以简单的理解为每个处理器同时处理的任务数量，也就是一次分配每个处理的任务数量。

ps:插一句，严格从执行顺序来看实际上每个batch内的原子任务还是顺序执行的（这点与深度学习框架中的batch是不同的），当每个任务执行时间非常短时，可以从逻辑上认为每个bacth内的任务是同时执行的。因此这个参数的应用场景是每个任务单独执行的时间非常短，但是又希望使用并行加快速度的场景。

n_jobs,pre_dispatch,batch_size这三个参数要综合理解。举个例子，假设我们有24个任务，设定n_jobs = 3(使用3个处理器)，pre_disaptch = 2 * 3，batch_size = 2

那么实际上预派遣了2*3*bacth_size=12个任务，也就是说队列里有12个任务可供随时选择执行。然后batch_size = 2,表示每个处理器需要拿走2个任务，n_jobs=3的情况下，总共需要拿走n_jobs*batch_szie = 6个任务去执行。这样队列里还有6个任务可供下次提取，保证了处理器不会空转。

• temp_folder: str, 可选项

这个看名字就知道是临时文件夹，存储进程中的一些临时缓存数据

• max_nbytes int, str, or None, optional, 1M by default

触发缓存机制的阈值，当worker中的numpy数组超过阈值时会触发内存映射

• mmap_mode: {None, 'r+', 'r', 'w+', 'c'}

映射后的数据的打开模式，就是常规的读写权限控制

• prefer: str in {'processes', 'threads'} or None, default: None

如果使用parallel_backend上下文管理器没有选择任何特定backend，则使用软提示选择默认backend。默认的基于进程(thread-based)的backend是“loky”，默认的基于线程的backend是“threading”。
如果指定了“backend”参数，则忽略。

• require: 'sharedmem' or None, default None

硬约束选择backend。如果设置为'sharedmem'，即使用户要求使用parallel_backend实现非基于线程的后端,所选backend也将是single-host和thread-based的。

使用示例：

from joblib import Parallel
#4个线程
works_4 = Parallel(n_jobs=4,backend = 'threading')
#2个进程
works_2 = Parallel(n_jobs=4,backend = 'multiprocessing')

（2）调用

Parallel实际上是一个类，但是实现了__call__方法，因此可以像函数一样进行调用。

但是Parallel的输入需要可迭代的对象，也就是一些任务的集合。这些任务由(function, args,kwargs)这种函数名和参数的方式表示。

from joblib import Parallel
def f(x,y):    
    return x+y
works_2 = Parallel(n_jobs=2,backend = 'threading')
task1 = [f, [2,3],{}]
task2 = [f,[4],{'y':5}]
res = works_2([task1,task2])
print(res)
#out: res = [5,9] 
 

上面的代码我们先定义了一个2线程的并行处理器works_2,然后构造了两个任务task1和task2。[task1,task2]合并成一个任务集合作为works_2的参数进行执行，最终我们也得到了2+3和4+5的正确结果。

我们可能会想任务集合中各个任务的函数可以不同吗？答案当然是可以的

from joblib import Parallel
def f1(x,y):    
    return x+y
def f2(x,y):    
    return y-x
task1 = [f1, [4,10],{}]
task2 = [f2,[4,10],{}]
works_2 = Parallel(n_jobs=2,backend = 'threading')
res = works_2([task1,task2])
print(res
#out: res = [14,6] 
 

讲到这儿就可以基本写出多任务并行的程序了，但是是不是觉得每次构造一个任务集合非常的麻烦，需要指定函数名，参数啥的。回想一下之前说的Dealyed函数不就是帮我们做这件事的吗，自动创建一个由(函数,args,kwargs)这样的原子组成的列表。

from joblib import delayed
def f(x,y):    
    return x+y
x_list = [1,2,3]
y_list = [-1,-2,-3]
tasks = [delayed(f)(x,y) for x,y in zip(x_list,y_list)]
print(tasks)

结果如下：

[(<function f at 0x00000000081C2950>, (1, -1), {}), 
(<function f at 0x00000000081C2950>, (2, -2), {}), 
(<function f at 0x00000000081C2950>, (3, -3), {})]

这下我们结合Parallel和delayed更为方便的创建并行程序

from joblib import Parallel,delayed
def f(x,y):    
    return x+y
x_list = [1,2,3]
y_list = [-1,-2,-3]
tasks = [delayed(f)(x,y) for x,y in zip(x_list,y_list)]    
print(tasks)
works_2 = Parallel(n_jobs=2,backend = 'threading')
res = works_2(tasks)
print(res)

为了程序简洁，我们可以省略一些中间过程，上面的代码也可以写为：

from joblib import Parallel,delayed
def f(x,y):    
    return x+y
x_list = [1,2,3]
y_list = [-1,-2,-3]
res = Parallel(n_jobs=2,backend = 'threading')([delayed(f)(x,y) for x,y in zip(x_list,y_list)] )  
print(res)

这就是我们一般看到的示例了。

到此这篇关于Python并行库joblib之delayed函数与Parallel函数详解的文章就介绍到这了,更多相关python并行库joblib内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！