Joblib

Joblib就是一個可以簡單地將Python代碼轉換為并行計算模式的軟件包，它可非常簡單并行我們的程序，從而提高計算速度。

主要提供了以下功能

• 程序并行
• 用于在每次使用相同的輸入參數調用函數時將其返回值緩存
• 數據存儲（包括不可哈希的數據和大規(guī)模numpy數組）

程序并行

joblib提供了一個簡單地程序并行方案，主要有Parallel函數實現(xiàn)，并涉及了一個技巧性的函數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函數顧名思義就是延遲函數的執(zhí)行。

根據源碼來看，delayed函數保留被修飾的函數function和參數*args, **kwargs，在碰到調用時，并不直接執(zhí)行函數function(*args, **kwargs),而是返回返回元組(function,args,kwargs)。

返回的這個結果留待其他函數執(zhí)行，在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)

執(zhí)行結果為：

(<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之后并未得到函數的執(zhí)行結果，我們如果想得到預期的執(zhí)行結果應該怎么做呢?其實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。

開始可能很難理解我們?yōu)槭裁匆舆t一個函數的調用呢，調用函數不就是為了得到執(zhí)行結果嗎，這樣延遲之后不就無法達到我們的預期了？

先別急，看看下面這個例子應該就明白了

假設我們需要對x_list = [1,2,3], y_list = [ -1,-2,-3] 這兩個列表進行逐個元素相加。

可能會覺得這不是很容易實現(xiàn)嗎，依次遍歷兩個集合的元素，在循環(huán)過程中調用函數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當然沒問題，但是如果我們并行的執(zhí)行f的三次調用，也就是同時執(zhí)行f(1,-1),f(2,-2),f(3,-3)時代碼該怎么寫呢？我們希望使用三個線程/進程，每個進程分別執(zhí)行一個函數調用。這時就可以看出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)

執(zhí)行之后的結果為：

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

我們獲得了由(函數,args,kwargs)這樣的元組組成的列表，我們可以為列表中的每一個元組(f,args,kwargs)分配給不同的線程，在每個線程里面執(zhí)行一個f(args,kwargs)。這樣是不是就可以完成并行的目的了呢?如何將元組分配給不同的線程由函數Parallel實現(xiàn)，后面再講。我們現(xiàn)在只需要知道的delayed函數只是為了生成(函數,args,kwargs)這樣的元組，暫緩函數的執(zhí)行，方便將各個計算過程分配給不同的線程。

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主要的功能是實現(xiàn)程序并行。

因此在講Parallel之前，先看一下并行的實際處理流程：

Parallel實際上就是封裝了這個任務拆分，并行和結果合并的這個過程。其主要的功能是線程和進程的創(chuàng)建，和多任務執(zhí)行這個流程。

由于其源碼過于復雜，這里只看和調用相關的兩個部分:

(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 —— 設置并行執(zhí)行任務的最大數量

當backend="multiprocessing"時指python工作進程的數量，或者backend="threading"時指線程池大小。
 當n_jobs=-1時，使用所有的CPU執(zhí)行并行計算；
當n_jobs=1時，就不會使用并行代碼，即等同于順序執(zhí)行，可以在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' —— 指定并行化后端的實現(xiàn)方法

backend='loky': 在與Python進程交換輸入和輸出數據時，可導致一些通信和內存開銷。
 
backend='multiprocessing': 基于multiprocessing.Pool的后端，魯棒性不如loky。
 
backend='threading': threading是一個開銷非常低的backend。但是如果被調用的函數大量依賴于Python對象，它就會受到Python全局解釋器(GIL)鎖的影響。當執(zhí)行瓶頸是顯式釋放GIL的已編譯擴展時，
“threading”非常有用(例如，封裝在“with nogil”塊中的Cython循環(huán)，或者對庫(如NumPy)的大量調用)。

• verbose: int, 可選項 —— 執(zhí)行期間的信息顯示（ps:這個參數實際上有點類似log的日志級別，數字越大，顯示的程序執(zhí)行過程越詳細）

信息級別:如果非零，則打印進度消息。超過50，輸出被發(fā)送到stdout。消息的頻率隨著信息級別的增加而增加。如果大于10，則報告所有迭代。

• timeout: float, 可選項 —— 任務運行時間限制

timeout僅用在n_jobs != 1的情況下，用來限制每個任務完成的時間，如果任何任務的執(zhí)行超過這個限制值，將會引發(fā)“TimeOutError”錯誤。

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

預先分派的(任務的)批數(batches)。默認設置是“2 * n_jobs”。

ps: 這個參數有點難理解，直觀來說一下吧。pre_dispatch是預派遣的意思，就是提前先把任務派遣給各個處理器。

注意一下，這里的派遣并不是口頭的安排任務，而是把任務和任務對應的數據(劃重點)也發(fā)送給處理器。假設我們總共有12，3個處理器，如果不設置pre_dispatch，那么程序在開始時會一次性將所有任務都分配出去，一個處理器領到了四個任務。但是這個派遣過程需要為每個任務準備相應的數據，需要處理時間和占據內存空間。 

一次分配任務過多可能造成的后果就是準備數據的時間變長和可能造成內存爆炸。所以設置一下預派遣任務的數量，減小預處理時間和內存占用，保證處理器不空轉就行。

• batch_size: int or 'auto', default: 'auto' —— 一次性分派給每個worker的原子任務的數量

當單個原子任務執(zhí)行非常快時，由于開銷的原因，使用dispatching的worker可能比順序計算慢。一起進行批量快速計算可以緩解這種情況。
 
“auto”策略會跟蹤一個批處理完成所需的時間，并動態(tài)調整batch_size大小，使用啟發(fā)式方法將時間保持在半秒以內。初始batch_size為1。
 
batch_size="auto"且backend="threading時，將一次分派一個任務的batches，因為threading后端有非常小的開銷，使用更大的batch_size在這種情況下沒有證明帶來任何好處。

這個可以簡單的理解為每個處理器同時處理的任務數量，也就是一次分配每個處理的任務數量。

ps:插一句，嚴格從執(zhí)行順序來看實際上每個batch內的原子任務還是順序執(zhí)行的（這點與深度學習框架中的batch是不同的），當每個任務執(zhí)行時間非常短時，可以從邏輯上認為每個bacth內的任務是同時執(zhí)行的。因此這個參數的應用場景是每個任務單獨執(zhí)行的時間非常短，但是又希望使用并行加快速度的場景。

n_jobs,pre_dispatch,batch_size這三個參數要綜合理解。舉個例子，假設我們有24個任務，設定n_jobs = 3(使用3個處理器)，pre_disaptch = 2 * 3，batch_size = 2

那么實際上預派遣了2*3*bacth_size=12個任務，也就是說隊列里有12個任務可供隨時選擇執(zhí)行。然后batch_size = 2,表示每個處理器需要拿走2個任務，n_jobs=3的情況下，總共需要拿走n_jobs*batch_szie = 6個任務去執(zhí)行。這樣隊列里還有6個任務可供下次提取，保證了處理器不會空轉。

• temp_folder: str, 可選項

這個看名字就知道是臨時文件夾，存儲進程中的一些臨時緩存數據

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

觸發(fā)緩存機制的閾值，當worker中的numpy數組超過閾值時會觸發(fā)內存映射

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

映射后的數據的打開模式，就是常規(guī)的讀寫權限控制

• 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實現(xiàn)非基于線程的后端,所選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實際上是一個類，但是實現(xiàn)了__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的參數進行執(zhí)行，最終我們也得到了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] 
 

講到這兒就可以基本寫出多任務并行的程序了，但是是不是覺得每次構造一個任務集合非常的麻煩，需要指定函數名，參數啥的?；叵胍幌轮罢f的Dealyed函數不就是幫我們做這件事的嗎，自動創(chuàng)建一個由(函數,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更為方便的創(chuàng)建并行程序

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內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！