python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python Joblib库总结

Python Joblib库使用方法案例总结

作者:授客

Python Joblib库是一个用于并行计算和数据预处理的工具库。它可以帮助用户快速处理大量数据,提高计算效率。其中,最常用的功能是并行计算,可以使用多个CPU核心同时处理任务,大大缩短计算时间。此外,Joblib还提供了一些数据预处理的功能,可以帮助用户更好地处理数据

实践环境

python 3.6.2

Joblib

简介

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

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

主要功能

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后端之外,还可以使用几个特定于集群的后端:

序列化与进程

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

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

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

共享内存语义

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 time
import threading
from joblib import Parallel, delayed, parallel_backend
from 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, delayed
from 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))

常用参数说明

参考文档

https://joblib.readthedocs.io/en/latest/

https://joblib.readthedocs.io/

https://joblib.readthedocs.io/en/latest/parallel.html#common-usage

Python Joblib库是一个非常实用的工具库,可以帮助用户更高效地处理数据。它提供了并行计算和数据预处理的功能,可以大大缩短计算时间,提高计算效率。使用Joblib库,用户只需要简单地调用相关函数即可完成任务,非常方便。在实际应用中,Joblib库可以帮助用户处理大量数据,提高数据处理的效率和准确性。因此,学习和掌握Joblib库的使用方法对于数据科学从业者来说非常重要。

到此这篇关于Python Joblib库使用方法案例总结的文章就介绍到这了,更多相关Python Joblib库总结内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文