# 查看版本
import joblib
joblib.__version__

class joblib.Memory(location=None, backend='local', mmap_mode=None, compress=False, verbose=1, bytes_limit=None, backend_options=None)

from joblib import Memory
import os, shutil
# 創(chuàng)建一個(gè)Memory對(duì)象，指定緩存目錄為當(dāng)前目錄下的run文件夾
# verbose=0表示關(guān)閉詳細(xì)輸出
cachedir = './run'
if os.path.exists(cachedir):
    shutil.rmtree(cachedir)
memory = Memory(cachedir, verbose=0)
# 使用@memory.cache裝飾器，將函數(shù)f的結(jié)果緩存起來(lái)
@memory.cache
def f(x):
    # 只有當(dāng)函數(shù)的輸入?yún)?shù)x沒有被緩存時(shí)，才會(huì)執(zhí)行函數(shù)體內(nèi)的代碼
    print('Running f(%s)' % x)
    return x
# 第一次調(diào)用f(1)，會(huì)執(zhí)行函數(shù)體內(nèi)的代碼，并將結(jié)果緩存起來(lái)
print(f(1))

Running f(1)
1

# 第二次調(diào)用f(1)，由于結(jié)果已經(jīng)被緩存，不會(huì)再次執(zhí)行函數(shù)體內(nèi)的代碼，而是直接從緩存中讀取結(jié)果
print(f(1))

# 調(diào)用f(2)，由于輸入?yún)?shù)不同，會(huì)再次執(zhí)行函數(shù)體內(nèi)的代碼，并將結(jié)果緩存起來(lái)
print(f(2))

Running f(2)
2

import numpy as np
from joblib import Memory
import os, shutil
cachedir = './run'
if os.path.exists(cachedir):
    shutil.rmtree(cachedir)
memory = Memory(cachedir, verbose=0)
@memory.cache
def g(x):
    print('A long-running calculation, with parameter %s' % x)
    # 返回漢明窗
    return np.hamming(x)
@memory.cache
def h(x):
    print('A second long-running calculation, using g(x)')
    # 生成范德蒙德矩陣
    return np.vander(x)
# 調(diào)用函數(shù)g，傳入?yún)?shù)3，并將結(jié)果存儲(chǔ)在變量a中
a = g(3)
# 打印變量a的值
print(a)
# 再次調(diào)用函數(shù)g，傳入相同的參數(shù)3，由于結(jié)果已被緩存，不會(huì)重新計(jì)算
print(g(3))

A long-running calculation, with parameter 3
[0.08 1.   0.08]
[0.08 1.   0.08]

# 調(diào)用函數(shù)h，傳入變量a作為參數(shù)，并將結(jié)果存儲(chǔ)在變量b中
b = h(a)
# 再次調(diào)用函數(shù)h，傳入相同的參數(shù)a，由于結(jié)果已被緩存，不會(huì)重新計(jì)算
b2 = h(a)
# 使用numpy的allclose函數(shù)檢查b和b2是否足夠接近，即它們是否相等
print(np.allclose(b, b2))

A second long-running calculation, using g(x)
True

import numpy as np
from joblib import Memory
import os, shutil
# 設(shè)置緩存目錄的路徑。
cachedir = './run'
# 檢查緩存目錄是否存在。
if os.path.exists(cachedir):
    # 如果緩存目錄存在，使用shutil.rmtree刪除該目錄及其內(nèi)容。
    shutil.rmtree(cachedir)
# 初始化Memory對(duì)象，設(shè)置緩存目錄為上面定義的cachedir，mmap_mode設(shè)置為'r'，表示只讀模式。
memory = Memory(cachedir, mmap_mode='r', verbose=0)
# 使用memory.cache裝飾器緩存np.square函數(shù)的結(jié)果。
square = memory.cache(np.square)
a = np.vander(np.arange(3)).astype(float)
# 打印通過square函數(shù)處理后的矩陣a。
print(square(a))
# 獲取a的緩存結(jié)果
result = square.call_and_shelve(a)
print(result.get())  # 獲取并打印緩存的結(jié)果。

[[ 0.  0.  1.]
 [ 1.  1.  1.]
 [16.  4.  1.]]
[[ 0.  0.  1.]
 [ 1.  1.  1.]
 [16.  4.  1.]]

@memory.cache
def compute_func(arg1, arg2, arg3):
    pass
class Foo(object):
    def __init__(self, args):
        self.data = None
    def compute(self):
        # 類中調(diào)用緩存的函數(shù)
        self.data = compute_func(self.arg1, self.arg2, 40)

class joblib.Parallel(n_jobs=default(None), backend=default(None), return_as='list', verbose=default(0), timeout=None, batch_size='auto', pre_dispatch='2 * n_jobs', temp_folder=default(None), max_nbytes=default('1M'), require=default(None))

from joblib import Parallel, delayed
import numpy as np
import time
start = time.time()
# 定義一個(gè)計(jì)算密集型函數(shù)
def compute_heavy_task(data):
    # 模擬處理時(shí)間
    time.sleep(1)
    # 數(shù)值計(jì)算
    result = np.sum(np.square(data))
    return result
# 生成一些模擬數(shù)據(jù)
# 設(shè)置隨機(jī)數(shù)生成器的種子
np.random.seed(42) 
data = np.random.rand(10, 1000)  # 10個(gè)1000維的向量
results = [compute_heavy_task(d) for d in data]
# 打印結(jié)果的和
print(f"結(jié)果: {sum(results)}")
print(f"耗時(shí)：{time.time()-start}s")

from joblib import Parallel, delayed
import numpy as np
import time
start = time.time()
# 定義一個(gè)計(jì)算密集型函數(shù)
def compute_heavy_task(data):
    # 模擬處理時(shí)間
    time.sleep(1)
    # 數(shù)值計(jì)算
    result = np.sum(np.square(data))
    return result
# 設(shè)置隨機(jī)數(shù)生成器的種子
np.random.seed(42) 
# 生成一些模擬數(shù)據(jù)
data = np.random.rand(10, 1000)  # 10個(gè)1000維的向量
# 使用Parallel來(lái)并行執(zhí)行任務(wù)
results = Parallel(n_jobs=8, return_as="generator")(delayed(compute_heavy_task)(d) for d in data)
# 打印結(jié)果的和
print(f"結(jié)果: {sum(results)}")
print(f"耗時(shí)：{time.time()-start}s")

import joblib
# 獲取當(dāng)前系統(tǒng)的cpu核心數(shù)
n_cores = joblib.cpu_count()
print(f'系統(tǒng)的核心數(shù)是：{n_cores}')

# 使用loky多進(jìn)程
from joblib import Parallel, delayed
import numpy as np
import time
start = time.time()
# 定義一個(gè)計(jì)算密集型函數(shù)
def compute_heavy_task(data):
    # 模擬處理時(shí)間
    time.sleep(1)
    # 數(shù)值計(jì)算
    result = np.sum(np.square(data))
    return result
# 生成一些模擬數(shù)據(jù)
data = np.random.rand(10, 1000)  # 10個(gè)1000維的向量
results = Parallel(n_jobs=8, return_as="generator", backend='loky')(delayed(compute_heavy_task)(d) for d in data)
# 打印結(jié)果的和
print(f"結(jié)果: {sum(results)}")
print(f"耗時(shí)：{time.time()-start}s")

# 使用threading多線程
start = time.time()
results = Parallel(n_jobs=8, return_as="generator", backend = 'threading')(delayed(compute_heavy_task)(d) for d in data)
# 打印結(jié)果的和
print(f"結(jié)果: {sum(results)}")
print(f"耗時(shí)：{time.time()-start}s")

from joblib import Parallel, delayed
import numpy as np
import time
# 定義一個(gè)計(jì)算密集型函數(shù)
def compute_heavy_task(data):
    # 模擬處理時(shí)間
    time.sleep(1)
    # 數(shù)值計(jì)算
    result = np.sum(np.square(data))
    return result
def main():
    start = time.time()
    # 生成一些模擬數(shù)據(jù)
    data = np.random.rand(10, 1000)  # 10個(gè)1000維的向量
    # multiprocessing不支持返回rgenerator
    results = Parallel(n_jobs=8, return_as="list",  backend='multiprocessing')(delayed(compute_heavy_task)(d) for d in data)
    # 打印結(jié)果的和
    print(f"結(jié)果: {sum(results)}")
    print(f"耗時(shí)：{time.time()-start}s")
if __name__ == '__main__':
    main()

from joblib import Parallel, delayed
shared_set = set()
def collect(x):
   shared_set.add(x)
Parallel(n_jobs=2)(delayed(collect)(i) for i in range(5))
print(sorted(shared_set))

# require='sharedmem'表示需要共享內(nèi)存，以確保多個(gè)進(jìn)程可以訪問shared_set集合
Parallel(n_jobs=2, require='sharedmem')(delayed(collect)(i) for i in range(5))
print(sorted(shared_set))

from joblib import Parallel, delayed
import math
with Parallel(n_jobs=2) as parallel:
   accumulator = 0.
   n_iter = 0
   while accumulator < 1000:
       results = parallel(delayed(math.sqrt)(accumulator + i ** 2) for i in range(5))
       accumulator += sum(results)
       n_iter += 1
print(accumulator, n_iter)  

from tempfile import mkdtemp
# 使用mkdtemp創(chuàng)建一個(gè)臨時(shí)目錄，并將目錄路徑存儲(chǔ)在變量savedir中。
savedir = mkdtemp(dir='./')
import os
# 文件保存路徑
filename = os.path.join(savedir, 'test.joblib')
import numpy as np
import pandas as pd
import joblib
# 創(chuàng)建一個(gè)要持久化的字典
to_persist = [('a', [1, 2, 3]), ('b', np.arange(10)), ('c', pd.DataFrame(np.ones((5,5))))]
# 使用joblib.dump函數(shù)將to_persist字典序列化并保存到filename指定的文件中
# 注意pickle庫(kù)無(wú)法序列化numpy數(shù)據(jù)
joblib.dump(to_persist, filename)

[('a', [1, 2, 3]),
 ('b', array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])),
 ('c',
       0    1    2    3    4
  0  1.0  1.0  1.0  1.0  1.0
  1  1.0  1.0  1.0  1.0  1.0
  2  1.0  1.0  1.0  1.0  1.0
  3  1.0  1.0  1.0  1.0  1.0
  4  1.0  1.0  1.0  1.0  1.0)]

with open(filename, 'wb') as fo:  
    # 使用joblib將對(duì)象to_persist序列化并寫入文件
   joblib.dump(to_persist, fo)
with open(filename, 'rb') as fo:  
   joblib.load(fo)

# compress參數(shù)為壓縮級(jí)別，取值為0到9，值越大壓縮效果越好。為0時(shí)表示不壓縮，默認(rèn)值為0
joblib.dump(to_persist, filename, compress=1)

['./tmp82ms1z5w\\test.joblib']

joblib.dump(to_persist, filename + '.gz', compress=('gzip', 3))  
joblib.load(filename + '.gz')
joblib.dump(to_persist, filename + '.bz2', compress=('bz2', 5))  
joblib.load(filename + '.bz2')

[('a', [1, 2, 3]),
 ('b', array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])),
 ('c',
       0    1    2    3    4
  0  1.0  1.0  1.0  1.0  1.0
  1  1.0  1.0  1.0  1.0  1.0
  2  1.0  1.0  1.0  1.0  1.0
  3  1.0  1.0  1.0  1.0  1.0
  4  1.0  1.0  1.0  1.0  1.0)]

joblib.dump(to_persist, filename, compress=('lz4', 3))  

['./tmp82ms1z5w\\test.joblib']

# 導(dǎo)入time模塊，用于實(shí)現(xiàn)延時(shí)功能
import time
# 定義一個(gè)模擬耗時(shí)計(jì)算的函數(shù)
def costly_compute(data, column):
    # 休眠1秒，模擬耗時(shí)操作
    time.sleep(1)
    # 返回傳入數(shù)據(jù)的指定列
    return data[column]
# 定義一個(gè)計(jì)算數(shù)據(jù)列平均值的函數(shù)
def data_processing_mean(data, column):
    # 調(diào)用costly_compute函數(shù)獲取指定列的數(shù)據(jù)
    column_data = costly_compute(data, column)
    # 計(jì)算并返回該列數(shù)據(jù)的平均值
    return column_data.mean()
# 導(dǎo)入numpy庫(kù)，并設(shè)置隨機(jī)數(shù)生成器的種子，以保證結(jié)果的可復(fù)現(xiàn)性
import numpy as np
rng = np.random.RandomState(42)
# 生成1000行4列的隨機(jī)數(shù)據(jù)矩陣
data = rng.randn(int(1000), 4)
# 記錄開始時(shí)間
start = time.time()
# 對(duì)數(shù)據(jù)的每一列計(jì)算平均值，并將結(jié)果存儲(chǔ)在results列表中
results = [data_processing_mean(data, col) for col in range(data.shape[1])]
# 記錄結(jié)束時(shí)間
stop = time.time()
# 打印處理過程的描述信息
print('\nSequential processing')
# 打印整個(gè)處理過程的耗時(shí)
print('Elapsed time for the entire processing: {:.2f} s'.format(stop - start))

Sequential processing
Elapsed time for the entire processing: 4.05 s

# 導(dǎo)入time模塊，用于模擬耗時(shí)操作。
import time
# 定義一個(gè)使用緩存的函數(shù)，用于計(jì)算數(shù)據(jù)的均值。
def data_processing_mean_using_cache(data, column):
    return costly_compute_cached(data, column).mean()
# 從joblib庫(kù)導(dǎo)入Memory類，用于緩存函數(shù)的輸出。
from joblib import Memory
# 設(shè)置緩存的存儲(chǔ)位置和詳細(xì)程度
location = './cachedir'
memory = Memory(location, verbose=0)
# 使用Memory對(duì)象的cache方法來(lái)緩存costly_compute函數(shù)的輸出。
costly_compute_cached = memory.cache(costly_compute)
# 從joblib庫(kù)導(dǎo)入Parallel和delayed類，用于并行執(zhí)行函數(shù)。
from joblib import Parallel, delayed
# 記錄開始時(shí)間。
start = time.time()
# 使用Parallel類并行執(zhí)行data_processing_mean_using_cache函數(shù)，對(duì)數(shù)據(jù)的每一列進(jìn)行處理。
results = Parallel(n_jobs=2)(
    delayed(data_processing_mean_using_cache)(data, col)  
    for col in range(data.shape[1])) 
# 記錄結(jié)束時(shí)間。
stop = time.time()
# 打印第一輪處理的耗時(shí)信息，包括緩存數(shù)據(jù)的時(shí)間。
print('\nFirst round - caching the data')
print('Elapsed time for the entire processing: {:.2f} s'.format(stop - start))

First round - caching the data
Elapsed time for the entire processing: 2.05 s

start = time.time()
results = Parallel(n_jobs=2)(
    delayed(data_processing_mean_using_cache)(data, col)
    for col in range(data.shape[1]))
stop = time.time()
print('\nSecond round - reloading from the cache')
print('Elapsed time for the entire processing: {:.2f} s'.format(stop - start))
# 如果不想使用緩存結(jié)果，可以清除緩存信息
memory.clear(warn=False)

Second round - reloading from the cache
Elapsed time for the entire processing: 0.02 s

import numpy as np
data = np.random.random((int(1e7),))
window_size = int(5e5)
slices = [slice(start, start + window_size)
          for start in range(0, data.size - window_size, int(1e5))]
import time
def slow_mean(data, sl):
    time.sleep(0.01)
    return data[sl].mean()

tic = time.time()
results = [slow_mean(data, sl) for sl in slices]
toc = time.time()
print('\nElapsed time computing the average of couple of slices {:.2f} s'.format(toc - tic))

Elapsed time computing the average of couple of slices 1.49 s

from joblib import Parallel, delayed
tic = time.time()
results = Parallel(n_jobs=2)(delayed(slow_mean)(data, sl) for sl in slices)
toc = time.time()
print('\nElapsed time computing the average of couple of slices {:.2f} s'.format(toc - tic))

Elapsed time computing the average of couple of slices 1.00 s

import os 
from joblib import dump, load  # 從joblib庫(kù)導(dǎo)入dump和load函數(shù)，用于創(chuàng)建和加載內(nèi)存映射文件
# 設(shè)置內(nèi)存映射文件的文件夾路徑
folder = './memmap'
os.makedirs(folder, exist_ok = True)
# 將內(nèi)存映射文件的名稱與路徑結(jié)合
data_filename_memmap = os.path.join(folder, 'data_memmap.joblib')
# 使用dump函數(shù)將數(shù)據(jù)對(duì)象'data'保存到內(nèi)存映射文件
dump(data, data_filename_memmap)
# 使用load函數(shù)加載內(nèi)存映射文件，mmap_mode='r'表示以只讀模式打開
data_ = load(data_filename_memmap, mmap_mode='r')
# 記錄開始時(shí)間
tic = time.time()
results = Parallel(n_jobs=2)(delayed(slow_mean)(data_, sl) for sl in slices)
# 記錄結(jié)束時(shí)間
toc = time.time()
print('\nElapsed time computing the average of couple of slices {:.2f} s\n'.format(toc - tic))  
import shutil
# 結(jié)束時(shí)刪除映射文件
try:
    shutil.rmtree(folder)
except: 
    pass

Elapsed time computing the average of couple of slices 0.77 s

from psutil import Process
from threading import Thread
class MemoryMonitor(Thread):
    """在單獨(dú)的線程中監(jiān)控內(nèi)存使用情況（以MB為單位）。"""
    def __init__(self):
        super().__init__()  # 調(diào)用父類Thread的構(gòu)造函數(shù)
        self.stop = False  # 用于控制線程停止的標(biāo)記
        self.memory_buffer = []  # 用于存儲(chǔ)內(nèi)存使用記錄的列表
        self.start()  # 啟動(dòng)線程
    def get_memory(self):
        """獲取進(jìn)程及其子進(jìn)程的內(nèi)存使用情況。"""
        p = Process()  # 獲取當(dāng)前進(jìn)程
        memory = p.memory_info().rss  # 獲取當(dāng)前進(jìn)程的內(nèi)存使用量
        for c in p.children():  # 遍歷所有子進(jìn)程
            memory += c.memory_info().rss  # 累加子進(jìn)程的內(nèi)存使用量
        return memory
    def run(self):
        """線程運(yùn)行的主體方法，周期性地記錄內(nèi)存使用情況。"""
        memory_start = self.get_memory()  # 獲取初始內(nèi)存使用量
        while not self.stop:  # 當(dāng)未設(shè)置停止標(biāo)記時(shí)循環(huán)
            self.memory_buffer.append(self.get_memory() - memory_start)  # 記錄當(dāng)前內(nèi)存使用量與初始內(nèi)存使用量的差值
            time.sleep(0.2)  # 休眠0.2秒
    def join(self):
        """重寫join方法，設(shè)置停止標(biāo)記并等待線程結(jié)束。"""
        self.stop = True  # 設(shè)置停止標(biāo)記
        super().join()  # 調(diào)用父類方法等待線程結(jié)束

import time
import numpy as np
def return_big_object(i):
    """生成并返回一個(gè)大型NumPy數(shù)組對(duì)象。"""
    time.sleep(.1)  # 休眠0.1秒模擬耗時(shí)操作
    return i * np.ones((10000, 200), dtype=np.float64) 
def accumulator_sum(generator):
    """累加生成器生成的所有值，并打印進(jìn)度。"""
    result = 0
    for value in generator:
        result += value
        # print(".", end="", flush=True)  # 打印點(diǎn)號(hào)并刷新輸出
    # print("")  # 打印換行符
    return result
from joblib import Parallel, delayed
monitor = MemoryMonitor()  # 創(chuàng)建內(nèi)存監(jiān)控器實(shí)例，并啟動(dòng)監(jiān)視
print("Running tasks with return_as='list'...")  # 打印啟動(dòng)任務(wù)信息
res = Parallel(n_jobs=2, return_as="list")(
    delayed(return_big_object)(i) for i in range(150)  # 使用joblib的Parallel功能并行執(zhí)行任務(wù)
)
res = accumulator_sum(res)  # 累加結(jié)果
print('All tasks completed and reduced successfully.')  # 打印任務(wù)完成信息
# 報(bào)告內(nèi)存使用情況
del res  # 清理結(jié)果以避免內(nèi)存邊界效應(yīng)
monitor.join()  # 等待內(nèi)存監(jiān)控線程結(jié)束
peak = max(monitor.memory_buffer) / 1e9  # 計(jì)算峰值內(nèi)存使用量，并轉(zhuǎn)換為GB
print(f"Peak memory usage: {peak:.2f}GB")  # 打印峰值內(nèi)存使用量

Running tasks with return_as='list'...
All tasks completed and reduced successfully.
Peak memory usage: 2.44GB

monitor_gen = MemoryMonitor()  # 創(chuàng)建內(nèi)存監(jiān)控器實(shí)例，并啟動(dòng)監(jiān)視
print("Running tasks with return_as='generator'...")  # 打印啟動(dòng)任務(wù)信息
res = Parallel(n_jobs=2, return_as="generator")(
    delayed(return_big_object)(i) for i in range(150)  
)
res = accumulator_sum(res)  # 累加結(jié)果
print('All tasks completed and reduced successfully.')  # 打印任務(wù)完成信息
# 報(bào)告內(nèi)存使用情況
del res  # 清理結(jié)果以避免內(nèi)存邊界效應(yīng)
monitor_gen.join()  # 等待內(nèi)存監(jiān)控線程結(jié)束
peak = max(monitor_gen.memory_buffer) / 1e9  # 計(jì)算峰值內(nèi)存使用量，并轉(zhuǎn)換為GB
print(f"Peak memory usage: {peak:.2f}GB")  # 打印峰值內(nèi)存使用量

Running tasks with return_as='generator'...
All tasks completed and reduced successfully.
Peak memory usage: 0.19GB

import matplotlib.pyplot as plt
plt.figure(0)
plt.semilogy(
    np.maximum.accumulate(monitor.memory_buffer),
    label='return_as="list"'
)
plt.semilogy(
    np.maximum.accumulate(monitor_gen.memory_buffer),
    label='return_as="generator"'
)
plt.xlabel("Time")
plt.xticks([], [])
plt.ylabel("Memory usage")
plt.yticks([1e7, 1e8, 1e9], ['10MB', '100MB', '1GB'])
plt.legend()
plt.show()

# 創(chuàng)建一個(gè)每個(gè)任務(wù)耗時(shí)可能不同的處理函數(shù)
def return_big_object_delayed(i):
    if (i + 20) % 60:
        time.sleep(0.1)
    else:
        time.sleep(5)
    return i * np.ones((10000, 200), dtype=np.float64)

monitor_delayed_gen = MemoryMonitor()
print("Create result generator on delayed tasks with return_as='generator'...")
res = Parallel(n_jobs=2, return_as="generator")(
    delayed(return_big_object_delayed)(i) for i in range(150)
)
res = accumulator_sum(res)
print('All tasks completed and reduced successfully.')
del res 
monitor_delayed_gen.join()
peak = max(monitor_delayed_gen.memory_buffer) / 1e6
print(f"Peak memory usage: {peak:.2f}MB")

Create result generator on delayed tasks with return_as='generator'...
All tasks completed and reduced successfully.
Peak memory usage: 784.23MB

monitor_delayed_gen_unordered = MemoryMonitor()
print(
  "Create result generator on delayed tasks with "
  "return_as='generator_unordered'..."
)
res = Parallel(n_jobs=2, return_as="generator_unordered")(
    delayed(return_big_object_delayed)(i) for i in range(150)
)
res = accumulator_sum(res)
print('All tasks completed and reduced successfully.')
del res 
monitor_delayed_gen_unordered.join()
peak = max(monitor_delayed_gen_unordered.memory_buffer) / 1e6
print(f"Peak memory usage: {peak:.2f}MB")

Create result generator on delayed tasks with return_as='generator_unordered'...
All tasks completed and reduced successfully.
Peak memory usage: 175.22MB

plt.figure(1)
plt.semilogy(
    np.maximum.accumulate(monitor_delayed_gen.memory_buffer),
    label='return_as="generator"'
)
plt.semilogy(
    np.maximum.accumulate(monitor_delayed_gen_unordered.memory_buffer),
    label='return_as="generator_unordered"'
)
plt.xlabel("Time")
plt.xticks([], [])
plt.ylabel("Memory usage")
plt.yticks([1e7, 1e8, 1e9], ['10MB', '100MB', '1GB'])
plt.legend()
plt.show()