Numpy Arrays 比较和索引

Question

我有 2 个大小不等的数组：

>>> np.size(array1)
4004001
>>> np.size(array2)
1000 

现在，需要将array2 中的每个元素与array1 中的所有元素进行比较，以找到与array2 中该元素的值最接近的元素。 找到这个值后，我需要将它存储在一个大小为 1000 的不同数组中 - 一个对应于 array2 的大小。

执行此操作的繁琐而粗略的方法可能是使用 for 循环并从数组 2 中获取每个元素，从数组 1 元素中减去其绝对值，然后取最小值 - 这将使我的代码非常慢。

我想使用 numpy 矢量化操作来执行此操作，但我有点碰壁了。

Answer 1

为了充分利用numpy 并行性，我们需要向量化函数。此外，所有值都使用相同的标准（最近）在相同的数组 (array1) 中找到。因此，可以专门制作一个专门在array1中查找的函数。

但是，要使解决方案更具可重用性，最好先制定一个更通用的解决方案，然后再将其转换为更具体的解决方案。因此，作为寻找最接近值的一般方法，我们从this find nearest solution 开始。然后我们把它变成一个更具体的向量化它，让它同时处理多个元素：

import math
import numpy as np
from functools import partial

def find_nearest_sorted(array,value):
    idx = np.searchsorted(array, value, side="left")
    if idx > 0 and (idx == len(array) or math.fabs(value - array[idx-1]) < math.fabs(value - array[idx])):
        return array[idx-1]
    else:
        return array[idx]

array1 = np.random.rand(4004001)
array2 = np.random.rand(1000)

array1_sorted = np.sort(array1)

# Partially apply array1 to find function, to turn the general function
# into a specific, working with array1 only.
find_nearest_in_array1 = partial(find_nearest_sorted, array1_sorted)

# Vectorize specific function to allow us to apply it to all elements of
# array2, the numpy way.
vectorized_find = np.vectorize(find_nearest_in_array1)

output = vectorized_find(array2)

希望这是您想要的，一个新向量，将array2 中的数据映射到array1 中最接近的值。

Answer 2

最“numpythonic”的方式是使用broadcasting。这是计算距离矩阵的一种快速简便的方法，然后您可以为其取绝对值的argmin。

array1 = np.random.rand(4004001)
array2 = np.random.rand(1000)

# Calculate distance matrix (on truncated array1 for memory reasons)
dmat = array1[:400400] - array2[:,None]

# Take the abs of the distance matrix and work out the argmin along the  last axis
ix = np.abs(dmat).argmin(axis=1)

dmat的形状：

(1000, 400400)

ix 的形状和内容：

(1000,)    
array([237473, 166831,  72369,  11663,  22998,  85179, 231702, 322752, ...])

但是，如果您一次性执行此操作会占用大量内存，并且对于您指定的数组大小实际上无法在我的 8GB 机器上运行，这就是我减小 array1 大小的原因。

要使其在内存限制下工作，只需将其中一个数组分割成块，然后依次对每个块应用广播（或并行化）。在这种情况下，我将array2 切成了 10 个块：

# Define number of chunks and calculate chunk size
n_chunks = 10
chunk_len = array2.size // n_chunks

# Preallocate output array
out = np.zeros(1000)

for i in range(n_chunks):
    s = slice(i*chunk_len, (i+1)*chunk_len)
    out[s] = np.abs(array1 - array2[s, None]).argmin(axis=1)

Answer 3

import numpy as np
a = np.random.random(size=4004001).astype(np.float16)
b = np.random.random(size=1000).astype(np.float16)
#use numpy broadcasting to compare pairwise difference and then find the min arg in a for each element in b. Finally extract elements from a using the argmin array as indexes. 
output = a[np.argmin(np.abs(b[:,None] -a),axis=1)]

此解决方案虽然简单，但可能会占用大量内存。如果在大型阵列上使用它可能需要进一步优化。