# 查看版本
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)建一個Memory對象，指定緩存目錄為當前目錄下的run文件夾
# verbose=0表示關閉詳細輸出
cachedir = './run'
if os.path.exists(cachedir):
    shutil.rmtree(cachedir)
memory = Memory(cachedir, verbose=0)
# 使用@memory.cache裝飾器，將函數(shù)f的結果緩存起來
@memory.cache
def f(x):
    # 只有當函數(shù)的輸入?yún)?shù)x沒有被緩存時，才會執(zhí)行函數(shù)體內的代碼
    print('Running f(%s)' % x)
    return x
# 第一次調用f(1)，會執(zhí)行函數(shù)體內的代碼，并將結果緩存起來
print(f(1))

Running f(1)
1

# 第二次調用f(1)，由于結果已經被緩存，不會再次執(zhí)行函數(shù)體內的代碼，而是直接從緩存中讀取結果
print(f(1))

# 調用f(2)，由于輸入?yún)?shù)不同，會再次執(zhí)行函數(shù)體內的代碼，并將結果緩存起來
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)
# 調用函數(shù)g，傳入?yún)?shù)3，并將結果存儲在變量a中
a = g(3)
# 打印變量a的值
print(a)
# 再次調用函數(shù)g，傳入相同的參數(shù)3，由于結果已被緩存，不會重新計算
print(g(3))

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

# 調用函數(shù)h，傳入變量a作為參數(shù)，并將結果存儲在變量b中
b = h(a)
# 再次調用函數(shù)h，傳入相同的參數(shù)a，由于結果已被緩存，不會重新計算
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
# 設置緩存目錄的路徑。
cachedir = './run'
# 檢查緩存目錄是否存在。
if os.path.exists(cachedir):
    # 如果緩存目錄存在，使用shutil.rmtree刪除該目錄及其內容。
    shutil.rmtree(cachedir)
# 初始化Memory對象，設置緩存目錄為上面定義的cachedir，mmap_mode設置為'r'，表示只讀模式。
memory = Memory(cachedir, mmap_mode='r', verbose=0)
# 使用memory.cache裝飾器緩存np.square函數(shù)的結果。
square = memory.cache(np.square)
a = np.vander(np.arange(3)).astype(float)
# 打印通過square函數(shù)處理后的矩陣a。
print(square(a))
# 獲取a的緩存結果
result = square.call_and_shelve(a)
print(result.get())  # 獲取并打印緩存的結果。

[[ 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):
        # 類中調用緩存的函數(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()
# 定義一個計算密集型函數(shù)
def compute_heavy_task(data):
    # 模擬處理時間
    time.sleep(1)
    # 數(shù)值計算
    result = np.sum(np.square(data))
    return result
# 生成一些模擬數(shù)據(jù)
# 設置隨機數(shù)生成器的種子
np.random.seed(42) 
data = np.random.rand(10, 1000)  # 10個1000維的向量
results = [compute_heavy_task(d) for d in data]
# 打印結果的和
print(f"結果: {sum(results)}")
print(f"耗時：{time.time()-start}s")

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

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

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

from joblib import Parallel, delayed
import numpy as np
import time
# 定義一個計算密集型函數(shù)
def compute_heavy_task(data):
    # 模擬處理時間
    time.sleep(1)
    # 數(shù)值計算
    result = np.sum(np.square(data))
    return result
def main():
    start = time.time()
    # 生成一些模擬數(shù)據(jù)
    data = np.random.rand(10, 1000)  # 10個1000維的向量
    # multiprocessing不支持返回rgenerator
    results = Parallel(n_jobs=8, return_as="list",  backend='multiprocessing')(delayed(compute_heavy_task)(d) for d in data)
    # 打印結果的和
    print(f"結果: {sum(results)}")
    print(f"耗時：{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'表示需要共享內存，以確保多個進程可以訪問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)建一個臨時目錄，并將目錄路徑存儲在變量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)建一個要持久化的字典
to_persist = [('a', [1, 2, 3]), ('b', np.arange(10)), ('c', pd.DataFrame(np.ones((5,5))))]
# 使用joblib.dump函數(shù)將to_persist字典序列化并保存到filename指定的文件中
# 注意pickle庫無法序列化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將對象to_persist序列化并寫入文件
   joblib.dump(to_persist, fo)
with open(filename, 'rb') as fo:  
   joblib.load(fo)

# compress參數(shù)為壓縮級別，取值為0到9，值越大壓縮效果越好。為0時表示不壓縮，默認值為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']

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

Sequential processing
Elapsed time for the entire processing: 4.05 s

# 導入time模塊，用于模擬耗時操作。
import time
# 定義一個使用緩存的函數(shù)，用于計算數(shù)據(jù)的均值。
def data_processing_mean_using_cache(data, column):
    return costly_compute_cached(data, column).mean()
# 從joblib庫導入Memory類，用于緩存函數(shù)的輸出。
from joblib import Memory
# 設置緩存的存儲位置和詳細程度
location = './cachedir'
memory = Memory(location, verbose=0)
# 使用Memory對象的cache方法來緩存costly_compute函數(shù)的輸出。
costly_compute_cached = memory.cache(costly_compute)
# 從joblib庫導入Parallel和delayed類，用于并行執(zhí)行函數(shù)。
from joblib import Parallel, delayed
# 記錄開始時間。
start = time.time()
# 使用Parallel類并行執(zhí)行data_processing_mean_using_cache函數(shù)，對數(shù)據(jù)的每一列進行處理。
results = Parallel(n_jobs=2)(
    delayed(data_processing_mean_using_cache)(data, col)  
    for col in range(data.shape[1])) 
# 記錄結束時間。
stop = time.time()
# 打印第一輪處理的耗時信息，包括緩存數(shù)據(jù)的時間。
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))
# 如果不想使用緩存結果，可以清除緩存信息
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庫導入dump和load函數(shù)，用于創(chuàng)建和加載內存映射文件
# 設置內存映射文件的文件夾路徑
folder = './memmap'
os.makedirs(folder, exist_ok = True)
# 將內存映射文件的名稱與路徑結合
data_filename_memmap = os.path.join(folder, 'data_memmap.joblib')
# 使用dump函數(shù)將數(shù)據(jù)對象'data'保存到內存映射文件
dump(data, data_filename_memmap)
# 使用load函數(shù)加載內存映射文件，mmap_mode='r'表示以只讀模式打開
data_ = load(data_filename_memmap, mmap_mode='r')
# 記錄開始時間
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\n'.format(toc - tic))  
import shutil
# 結束時刪除映射文件
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):
    """在單獨的線程中監(jiān)控內存使用情況（以MB為單位）。"""
    def __init__(self):
        super().__init__()  # 調用父類Thread的構造函數(shù)
        self.stop = False  # 用于控制線程停止的標記
        self.memory_buffer = []  # 用于存儲內存使用記錄的列表
        self.start()  # 啟動線程
    def get_memory(self):
        """獲取進程及其子進程的內存使用情況。"""
        p = Process()  # 獲取當前進程
        memory = p.memory_info().rss  # 獲取當前進程的內存使用量
        for c in p.children():  # 遍歷所有子進程
            memory += c.memory_info().rss  # 累加子進程的內存使用量
        return memory
    def run(self):
        """線程運行的主體方法，周期性地記錄內存使用情況。"""
        memory_start = self.get_memory()  # 獲取初始內存使用量
        while not self.stop:  # 當未設置停止標記時循環(huán)
            self.memory_buffer.append(self.get_memory() - memory_start)  # 記錄當前內存使用量與初始內存使用量的差值
            time.sleep(0.2)  # 休眠0.2秒
    def join(self):
        """重寫join方法，設置停止標記并等待線程結束。"""
        self.stop = True  # 設置停止標記
        super().join()  # 調用父類方法等待線程結束

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

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

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

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)建一個每個任務耗時可能不同的處理函數(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()