在 numpy 矩阵上并行 for 循环

Question

我正在查看 joblib 示例，但我不知道如何在矩阵上执行并行 for 循环。我正在计算矩阵行之间的成对距离度量。所以我在做：

N, _ = data.shape
upper_triangle = [(i, j) for i in range(N) for j in range(i + 1, N)]
dist_mat = np.zeros((N,N))  

for (i, j) in upper_triangle:
    dist_mat[i,j] = dist_fun(data[i], data[j])
    dist_mat[j,i] = dist_mat[i,j]

其中 dist_fun 取两个向量并计算距离。我怎样才能使这个循环并行，因为对 dist_fun 的调用可以相互独立。

编辑：我使用的距离函数是fastdtw，不是那么快。所以我认为确实想要将其并行化。使用：

dist_mat = pdist(data, lambda x,y : fastdtw(x,y, dist=euclidean)[0])

我得到 58.1084 秒的执行时间，并使用：

dist_mat = np.zeros((N,N))
for (i,j), _ in np.ndenumerate(dist_mat):
    dist_mat[i,j], _ = fastdtw(data[i,:], timeseries[j,:], dist=euclidean)

我得到 116.36 秒并使用：

upper_triangle = [(i,j) for i in range(N) for j in range(i+1, N)]
dist_mat = np.zeros((N,N))

for (i,j) in upper_triangle:
    dist_mat[i,j], _ = fastdtw(data[i,:], data[j,:], dist=euclidean)
    dist_mat[j,i] = dist_mat[i,j]

我得到 55.62 秒。这里N=33。 scipy 是否自动使用所有可用内核？

编辑：我想我已经找到了使用 multiprocessing 包的解决方法，但我将不回答 joblib[=46= 的问题] 人在我 post 我认为有效之前做出回应。

Answer 1

这可以使用 multiprocessing 模块按如下方式完成：

import numpy as np
from fastdtw import fastdtw
import multiprocessing as mp
from scipy.spatial.distance import squareform, euclidean
from functools import partial

# Create simulated data matrix
data = np.random.random((33,300))

N, _ = data.shape
upper_triangle = [(i,j) for i in range(N) for j in range(i+1, N)]

with mp.Pool(processes=4) as pool:
    result = pool.starmap(partial(fastdtw, dist=euclidean), [(data[i], data[j]) for (i,j) in upper_triangle])

dist_mat = squareform([item[0] for item in result])

使用timeit on an IvyBridge Core-i5的计时结果：

24.052 secs

这是没有显式并行化的一半时间。

另外：

作为使用 fastdtw 包的任何人的未来参考。从 scipy.spatial.distance 导入距离函数并调用 fastdtw，如 link 中的示例所示，比仅使用 fastdtw(x,y,dist=2) 慢得多。结果似乎与我相似，使用 pdist（不求助于并行化）的执行时间不到一秒。

在 numpy 矩阵上并行 for 循环

Parallel for loop over numpy matrix

parallel-processing

python-3.x

joblib

python-multiprocessing

dtw