用途

  • 加速函数执行

    • 函数结果缓存
    • 第二次执行，直接返回缓存的结构，无需执行代码块
  • 并行运算

  • 替代 pickle

    • 更高效
  • 函数式编程

Memory 类

初始化

  • Memory(cachedir: str|None='your/given/path', verbose: int=0)

  • 参数

    • cachedir

      • 缓存路径

      • 取值

        • None: 不进行缓存

      • 路径

        • 存放缓存

使用 memmapping 加速

  • 参考：numpy.memmap

  • 参数

    • Memory(…, mmap_mode='r')

    • mmap_mode 取值

      • None: 默认值，不使用 memmapping 加速
      • 'r': 只读
      • 'r+' or 'w+': 会修改磁盘中存储的数据

      • 'c': copy on write，写时再复制一份在内存中

        • 即不会更改磁盘文件

  • 注意

    • 使用 memmapping 模式函数返回值（ndarray）

      • 是直接映射到 磁盘存储上的

    • 使用完返回值，要删除 del resturn_value

      • 避免文件锁异常

  • 例子

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

    import numpy as np
from joblib import Memory

memory = Memory('/tmp/hello', mmap_mode='r')

square = memory(np.square)


result = square(np.arange(3))

print(repr(result))

del result # 删除，防止文件锁，尤其是Windows

忽略参数列表

  • 说明

    • 设有两个参数（a,b），不想因为 b 不同，被再次执行
    • 即，只有 a 不同，才会触发函数的执行
    • 那么 参数 b 就是我们想忽略的参数

  • 使用： memory.cache(ignore=['b', 'debug', …])

    1
2
3
4
5
6
7
8

    >>> @memory.cache(ignore=['debug'])
... def my_func(x, debug=True):
...     print('Called with x = %s' % x)
>>> my_func(0)
Called with x = 0
>>> my_func(0, debug=False)
>>> my_func(0, debug=True)
>>> # my_func was not reevaluated

做装饰器使用

  • 例子

    1
2
3
4
5
6
7
8

    >>> import numpy as np

>>> @memory.cache
... def g(x):
...     print('A long-running calculation, with parameter %s' % x)
...     return np.hamming(x)

g(np.arange(3))

返回引用

  • 方法

    • 调用： refObject = cached_fun.call_and_shelve(*args, **kwargs)
    • 取回结果： refObject.get()

    • 删除磁盘缓存： refObject.clear()

      • 注意

        • clear 清理后，不能再回去结果，
        • refObject.get() 报 KeyError 异常

  • 优点

    • 降低传输空间消耗

      • eg: 不同 worker 间传递结果

注意和陷阱

识别符问题

  • Memory 跨 sesson 使用 function_name 辨识不同 函数

    • 要点 ：避免同名函数
    • 同一个 session 内还是能够分辨 同名函数的

    • 场景

      • 退出后，再次启动解释器 interpreter， 就不能分辨之前缓存的同名函数了

lambda 函数

  • python2.7 不能识别不同的 lambda 函数

不能用于 类实现的函数 __call__()

不能用于方法

  • 因为，被绑定的 obj, 是变化的

joblib 不同版本之间，cache 可能失效

并行计算

Parallel 类

  • 特点：

    • 已经设好的环境，直接使用

  • 例子

    1
2
3

    >>> 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]

  • 并行后台 backend

    • 如果不指定 backend 或 pefer

      • 使用默认值 'loky'

      • 若在 with parallel_backend 语句中

        • backend 由 parallel_backend('backend_name') 指定

parallel_backend 方法

  • 特点

    • with context 用法
    • 可以选择并行后台工具（loky, threading, multiprocessing, ray, …）

  • 例子

    • threading

      1
2
3
4

      >>> 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]

    • ray

      • 参考：Distributed Scikit-learn / Joblib — Ray v1.6.0

      • 例子

         1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

        import joblib
from ray.util.joblib import register_ray

# ray.init(...)
register_ray()
with joblib.parallel_backend('ray'):
    search.fit(digits.data, digits.target)

>>> from operator import neg
>>> from ray.util.joblib import register_ray
>>> register_ray()
>>> with parallel_backend("ray"):
...     print(Parallel()(delayed(neg)(i + 1) for i in range(5)))
[-1, -2, -3, -4, -5]