我正在解決一個匹配問題。我有兩個陣列A和B相同尺寸的(1000×1000承擔)。對于每一個元素a_ij中A(i和j是行和列索引,分別),我需要找到最接近的元素b_i?中i-th的行B。目前,我提出了三個解決方案:
- 嵌套回圈(耗時 5.47 秒)
- 帶廣播的單回圈(耗時 6.40 秒)
- 并行回圈(耗時 30.27 秒)
我認為上述三種方法的時間效率不夠高,或者至少我的實作還不足以實作這一點(第三種方法花費的時間比我預期的要長!)。如果您能指出如何改進我的代碼,那就太好了。提前致謝!
我目前的解決方案如下:
import numpy as np
import time
from joblib import Parallel, delayed
def method1(A,B): # Nested loop (as a Benchmark)
output = np.zeros(A.shape)
for r in range(A.shape[0]):
rowB = B[r]
for c in range(A.shape[1]):
elementA = A[r,c]
diff_vector = np.abs(elementA - rowB)
output[r,c] = np.argmin(diff_vector)
return output
def method2(A,B): # Broadcasting
output = np.zeros(A.shape)
for r in range(A.shape[0]):
diff_matrix = np.abs(A[r][:, np.newaxis] - B[r])
output[r] = np.argmin(diff_matrix, axis=1)
return output
def method3(A,B): # Parallel for loop
def matcher(r, A, B):
i = r//A.shape[1]
j = r % A.shape[1]
elementA = A[i, j]
rowB = B[i]
diff_vector = np.abs(elementA - rowB)
return np.argmin(diff_vector)
output = Parallel(n_jobs=4)(delayed(matcher)(r, A, B) for r in range(A.shape[0]*A.shape[1]))
output = np.reshape(output, [A.shape[0], A.shape[1]])
return output
A = np.random.randn(1000,1000)
B = np.random.randn(1000,1000)
output1 = method1(A,B)
output2 = method2(A,B)
output3 = method3(A,B)
uj5u.com熱心網友回復:
該演算法顯然效率低下(所有方法):檢查所有項rowB以找到最接近的一項是昂貴的,并且會導致數十億次浮點運算。更不用說 Numpy為每次呼叫和創建一個昂貴的不必要的臨時陣列。elementA - rowBnp.abs
您可以通過進行排序然后進行二分搜索來加快這一速度。這種方法在時間上的復雜性是O(n**2 log n),而對于初始方法來說是O(n**3)。您還可以使用Numba輕松撰寫此演算法的并行實作。此外,您不需要將輸出存盤在 float64 型別的陣列中:基于整數的陣列就足夠了(更快并且可能更小)。
這是結果實作:
import numba as nb
import numpy as np
@nb.njit('int_[:,::1](float64[:,::1],float64[:,::1])', parallel=True)
def method4(A,B):
mB = B.shape[1]
output = np.empty(A.shape, dtype=np.int_)
# Parallel loop
for r in nb.prange(A.shape[0]):
rowA = A[r]
rowB = B[r]
# Sort a row
index_rowB = np.argsort(rowB)
sorted_rowB = rowB[index_rowB]
# Fast binary search in the sorted row
# See: https://stackoverflow.com/a/46184652/12939557
idxs = np.searchsorted(sorted_rowB, rowA)
left = np.fabs(rowA - sorted_rowB[np.maximum(idxs-1, 0)])
right = np.fabs(rowA - sorted_rowB[np.minimum(idxs, mB-1)])
prev_idx_is_less = (idxs == mB) | (left < right)
# Find the position in the original unsorted array
output[r] = index_rowB[idxs - prev_idx_is_less]
return output
output4 = method4(A,B)
這比我的 10 核機器上最快的初始實作 ( )快 482 倍method1:
method1: 4341 ms
method2: 5839 ms
method4: 9 ms
轉載請註明出處,本文鏈接:https://www.uj5u.com/qukuanlian/404897.html
標籤:
上一篇:完成一個稀疏的時間線
