实践环境

python 3.6.2

Joblib

简介

Joblib 是一组在 Python 中提供轻量级流水线的工具。特别是：

1. 函数的透明磁盘缓存和延迟重新计算（记忆模式）

2. 简单易用的并行计算

Joblib 已被优化得很快速，很健壮了，特别是在大数据上，并对 numpy 数组进行了特定的优化。

主要功能

1.输出值的透明快速磁盘缓存(Transparent and fast disk-caching of output value): 

Python 函数的内存化或类似 make 的功能，适用于任意 Python 对象，包括非常大的 numpy 数组。通过将操作写成一组具有定义良好的输入和输出的步骤：Python 函数，将持久性和流执行逻辑与域逻辑或算法代码分离开来。Joblib 可以将其计算保存到磁盘上，并仅在必要时重新运行

原文：

Transparent and fast disk-caching of output value: a memoize or make-like functionality for Python functions that works well for arbitrary Python objects, including very large numpy arrays. Separate persistence and flow-execution logic from domain logic or algorithmic code by writing the operations as a set of steps with well-defined inputs and outputs: Python functions. Joblib can save their computation to disk and rerun it only if necessary:

>>> from joblib import Memory>>> cachedir = 'your_cache_dir_goes_here'>>> mem = Memory(cachedir)>>> import numpy as np>>> a = np.vander(np.arange(3)).astype(float)>>> square = mem.cache(np.square)>>> b = square(a)                                   ______________________________________________________________________...[Memory] Calling square...square(array([[0., 0., 1.],       [1., 1., 1.],       [4., 2., 1.]]))_________________________________________________...square - ...s, 0.0min
>>> c = square(a)# The above call did not trigger an evaluation

2.并行助手（parallel helper)

轻松编写可读的并行代码并快速调试

>>> from joblib import Parallel, delayed>>> from math import sqrt>>> Parallel(n_jobs=1)(delayed(sqrt)(i**2) for i in range(10))[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]
>>> res = Parallel(n_jobs=1)(delayed(sqrt)(i**2) for i in range(10))>>> res[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

3.快速压缩的持久化(Fast compressed Persistence)：

代替 pickle 在包含大数据的 Python 对象上高效工作（joblib.dump&joblib.load）。

parallel for loops

常见用法

Joblib 提供了一个简单的助手类，用于使用多进程为循环实现并行。核心思想是将要执行的代码编写为生成器表达式，并将其转换为并行计算

>>> from math import sqrt>>> [sqrt(i ** 2) for i in range(10)][0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

使用以下代码，可以分布到 2 个 CPU 上：

>>> from math import sqrt>>> from joblib import Parallel, delayed>>> Parallel(n_jobs=2)(delayed(sqrt)(i ** 2) for i in range(10))[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

输出可以是一个生成器，在可以获取结果时立即返回结果，即使后续任务尚未完成。输出的顺序始终与输入的顺序相匹配：输出的顺序总是匹配输入的顺序：

>>> from math import sqrt>>> from joblib import Parallel, delayed>>> parallel = Parallel(n_jobs=2, return_generator=True) # py3.7往后版本才支持return_generator参数>>> output_generator = parallel(delayed(sqrt)(i ** 2) for i in range(10))>>> print(type(output_generator))<class 'generator'>>>> print(next(output_generator))0.0>>> print(next(output_generator))1.0>>> print(list(output_generator))[2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

此生成器允许减少joblib.Parallel的内存占用调用

基于线程的并行 VS 基于进程的并行

默认情况下，joblib.Parallel使用'loky'后端模块启动单独的 Python 工作进程，以便在分散的 CPU 上同时执行任务。对于一般的 Python 程序来说，这是一个合理的默认值，但由于输入和输出数据需要在队列中序列化以便同工作进程进行通信，因此可能会导致大量开销（请参阅序列化和进程）。

当你知道你调用的函数是基于一个已编译的扩展，并且该扩展在大部分计算过程中释放了 Python 全局解释器锁（GIL）时，使用线程而不是 Python 进程作为并发工作者会更有效。例如，在 Cython 函数的with nogil 块中编写 CPU 密集型代码。

如果希望代码有效地使用线程，只需传递preferre='threads'作为joblib.Parallel构造函数的参数即可。在这种情况下，joblib 将自动使用"threading"后端，而不是默认的"loky"后端

>>> Parallel(n_jobs=2, prefer=threads')(...     delayed(sqrt)(i ** 2) for i in range(10))[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

也可以在上下文管理器的帮助下手动选择特定的后端实现:

>>> from joblib import parallel_backend>>> with parallel_backend('threading', n_jobs=2):...    Parallel()(delayed(sqrt)(i ** 2) for i in range(10))...[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

后者在调用内部使用joblib.Parallel的库时特别有用，不会将后端部分作为其公共 API 的一部分公开。

'loky'后端可能并不总是可获取。

一些罕见的系统不支持多处理（例如 Pyodide）。在这种情况下，loky 后端不可用，使用线程作为默认后端。

除了内置的 joblib 后端之外，还可以使用几个特定于集群的后端：

• 用于 Dask 集群的Dask后端 (查阅Using Dask for single-machine parallel computing 以获取示例),

• 用于 Ray 集群的Ray后端

• 用于 Spark 集群上分发 joblib 任务的Joblib Apache Spark Backend
  

序列化与进程

要在多个 python 进程之间共享函数定义，必须依赖序列化协议。python 中的标准协议是pickle ，但它在标准库中的默认实现有几个限制。例如，它不能序列化交互式定义的函数或在__main__模块中定义的函数。

为了避免这种限制，loky后端现在依赖于cloudpickle以序列化 python 对象。cloudpickle是pickle协议的另一种实现方式，允许序列化更多的对象，特别是交互式定义的函数。因此，对于大多数用途，loky后端应该可以完美的工作。

cloudpickle的主要缺点就是它可能比标准类库中的pickle慢，特别是，对于大型 python 字典或列表来说，这一点至关重要，因为它们的序列化时间可能慢 100 倍。有两种方法可以更改 joblib的序列化过程以缓和此问题：

• 如果您在 UNIX 系统上，则可以切换回旧的multiprocessing后端。有了这个后端，可以使用很快速的pickle在工作进程中共享交互式定义的函数。该解决方案的主要问题是，使用fork启动进程会破坏标准 POSIX，并可能与numpy和openblas等第三方库进行非正常交互。

• 如果希望将loky后端与不同的序列化库一起使用，则可以设置LOKY_PICKLER=mod_pickle环境变量，以使用mod_pickle作为loky的序列化库。作为参数传递的模块mod_pickle应按import mod_picke导入，并且应包含一个Pickler 对象，该对象将用于序列化为对象。可以设置LOKY_PICKLER=pickle以使用表中类库中的 pickling 模块。LOKY_PICKLER=pickle的主要缺点是不能序列化交互式定义的函数。为了解决该问题，可以将此解决方案与joblib.wrap_non_picklable_objects() 一起使用，joblib.wrap_non_picklable_objects()可用作装饰器以为特定对下本地启用cloudpickle。通过这种方式，可以为所有 python 对象使用速度快的 picking，并在本地为交互式函数启用慢速的 pickling。查阅loky_wrapper获取示例。

共享内存语义

joblib 的默认后端将在独立的 Python 进程中运行每个函数调用，因此它们不能更改主程序中定义的公共 Python 对象。

然而，如果并行函数确实需要依赖于线程的共享内存语义，则应显示的使用require='sharemem'，例如：

>>> shared_set = set()>>> def collect(x):...    shared_set.add(x)...>>> Parallel(n_jobs=2, require='sharedmem')(...     delayed(collect)(i) for i in range(5))[None, None, None, None, None]>>> sorted(shared_set)[0, 1, 2, 3, 4]

请记住，从性能的角度来看，依赖共享内存语义可能是次优的，因为对共享 Python 对象的并发访问将受到锁争用的影响。

注意，不使用共享内存的情况下，任务进程之间的内存资源是相互独立的，举例说明如下：

#!/usr/bin/env python# -*- coding:utf-8 -*-
import timeimport threadingfrom joblib import Parallel, delayed, parallel_backendfrom collections import deque
GLOBAL_LIST = []
class TestClass():    def __init__(self):        self.job_queue = deque()
    def add_jobs(self):        i = 0        while i < 3:            time.sleep(1)            i += 1            GLOBAL_LIST.append(i)            self.job_queue.append(i)            print('obj_id：', id(self),  'job_queue：', self.job_queue, 'global_list：', GLOBAL_LIST)

def get_job_queue_list(obj):    i = 0    while not obj.job_queue and i < 3:        time.sleep(1)        i += 1        print('obj_id：', id(obj), 'job_queue：', obj.job_queue, 'global_list：', GLOBAL_LIST)    return obj.job_queue

if __name__ == "__main__":    obj = TestClass()
    def test_fun():        with parallel_backend("multiprocessing", n_jobs=2):            Parallel()(delayed(get_job_queue_list)(obj) for i in range(2))
    thread = threading.Thread(target=test_fun, name="parse_log")    thread.start()
    time.sleep(1)    obj.add_jobs()    print('global_list_len：', len(GLOBAL_LIST))

控制台输出：

obj_id： 1554577912664 job_queue： deque([]) global_list： []obj_id： 1930069893920 job_queue： deque([]) global_list： []obj_id： 2378500766968 job_queue： deque([1]) global_list： [1]obj_id： 1554577912664 job_queue： deque([]) global_list： []obj_id： 1930069893920 job_queue： deque([]) global_list： []obj_id： 2378500766968 job_queue： deque([1, 2]) global_list： [1, 2]obj_id： 1554577912664 job_queue： deque([]) global_list： []obj_id： 1930069893920 job_queue： deque([]) global_list： []obj_id： 2378500766968 job_queue： deque([1, 2, 3]) global_list： [1, 2, 3]global_list_len： 3

通过输出可知，通过 joblib.Parallel 开启的进程，其占用内存和主线程占用的内存资源是相互独立

复用 worer 池

一些算法需要对并行函数进行多次连续调用，同时对中间结果进行处理。在一个循环中多次调用joblib.Parallel次优的，因为它会多次创建和销毁一个 workde（线程或进程）池，这可能会导致大量开销。

在这种情况下，使用joblib.Parallel类的上下文管理器 API 更有效，以便对joblib.Parallel对象的多次调用可以复用同一 worker 池。

from joblib import Parallel, delayedfrom math import sqrt
with Parallel(n_jobs=2) as parallel:   accumulator = 0.   n_iter = 0   while accumulator < 1000:       results = parallel(delayed(sqrt)(accumulator + i ** 2) for i in range(5))       accumulator += sum(results)  # synchronization barrier       n_iter += 1
print(accumulator, n_iter)  #输出： 1136.5969161564717 14                          

请注意，现在基于进程的并行默认使用'loky'后端，该后端会自动尝试自己维护和重用 worker 池，即使是在没有上下文管理器的调用中也是如此

笔者实践发现，即便采用这种实现方式，其运行效率也是非常低下的，应该尽量避免这种设计(实践环境 Python3.6)

Parallel 参考文档

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

常用参数说明

• n_jobs：int, 默认：None
  

  并发运行作业的最大数量，例如当backend='multiprocessing'时 Python 工作进程的数量，或者当backend='threading'时线程池的大小。如果设置为 -1，则使用所有 CPU。如果设置为 1，则根本不使用并行计算代码，并且行为相当于一个简单的 python for 循环。此模式与timeout不兼容。如果n_jobs小于-1，则使用(n_cpus+1+n_jobs)。因此，如果n_jobs=-2，将使用除一个 CPU 之外的所有 CPU。如果为None，则默认n_jobs=1，除非在parallel_backend()上下文管理器下执行调用，此时会为n_jobs设置另一个值。

• backend: str, ParallelBackendBase实例或者None, 默认: 'loky'
  

  指定并行化后端实现。支持的后端有：

  loky 在与工作 Python 进程交换输入和输出数据时，默认使用的loky可能会导致一些通信和内存开销。在一些罕见的系统（如 Pyiode）上，loky后端可能不可用。

  multiprocessing 以前基于进程的后端，基于multiprocessing.Pool。不如 loky 健壮。

  threading 是一个开销很低的后端，但如果被调用的函数大量依赖于 Python 对象，它会受到 Python GIL 的影响。当执行瓶颈是显式释放 GIL 的已编译扩展时，threading最有用（例如，with-nogil块中封装的 Cython 循环或对 NumPy 等库的昂贵调用）。

  最后，可以通过调用register_pallel_backend()来注册后端。

  不建议在类库中调用Parallel时对backend名称进行硬编码，取而代之，建议设置软提示（prefer）或硬约束（require），以便库用户可以使用parallel_backend()上下文管理器从外部更改backend。

• return_generator: bool

  如果为True，则对此实例的调用将返回一个生成器，并在结果可获取时立即按原始顺序返回结果。请注意，预期用途是一次运行一个调用。对同一个 Parallel 对象的多次调用将导致RuntimeError
  

• prefer: str 可选值 ‘processes’, ‘threads’ ,None, 默认: None
  

  如果使用parallel_backen()上下文管理器时没有指定特定后端，则选择默认prefer给定值。默认的基于进程的后端是loky，而默认的基于线程的后端则是threading。如果指定了backend参数，则忽略该参数。

• require: ‘sharedmem’ 或者None, 默认None
  

  用于选择后端的硬约束。如果设置为'sharedmem'，则所选后端将是单主机和基于线程的，即使用户要求使用具有parallel_backend的非基于线程的后端。

文章转载自：授客

原文链接：https://www.cnblogs.com/shouke/p/17472017.html