查找多个数组具有相同值的索引的快速算法

Question

我正在寻找加快（或替换）我的数据分组算法的方法。

我有一个 numpy 数组列表。我想生成一个新的 numpy 数组，这样该数组的每个元素对于原始数组也相同的每个索引都是相同的。 （在不是这种情况的情况下有所不同。）

这听起来有点尴尬，所以举个例子：

# Test values:
values = [
    np.array([10, 11, 10, 11, 10, 11, 10]),
    np.array([21, 21, 22, 22, 21, 22, 23]),
    ]

# Expected outcome: np.array([0, 1, 2, 3, 0, 3, 4])
#                             *           *

请注意，我标记的预期结果的元素（索引 0 和 4）具有相同的值（0），因为原始的两个数组也相同（即10 和21）。索引为 3 和 5 的元素类似 (3)。

该算法必须处理任意数量（大小相等）的输入数组，并且还为每个结果数返回它们对应的原始数组的值。 （所以对于这个例子，“3”指的是(11, 22)。）

这是我目前的算法：

import numpy as np

def groupify(values):
    group = np.zeros((len(values[0]),), dtype=np.int64) - 1 # Magic number: -1 means ungrouped.
    group_meanings = {}
    next_hash = 0
    matching = np.ones((len(values[0]),), dtype=bool)
    while any(group == -1):
        this_combo = {}

        matching[:] = (group == -1)
        first_ungrouped_idx = np.where(matching)[0][0]

        for curr_id, value_array in enumerate(values):
            needed_value = value_array[first_ungrouped_idx]
            matching[matching] = value_array[matching] == needed_value
            this_combo[curr_id] = needed_value
        # Assign all of the found elements to a new group
        group[matching] = next_hash
        group_meanings[next_hash] = this_combo
        next_hash += 1
    return group, group_meanings

请注意，表达式 value_array[matching] == needed_value 会针对每个单独的索引进行多次评估，这就是缓慢的原因。

我不确定我的算法是否可以进一步加快速度，但我也不确定它是否是开始时的最佳算法。有更好的方法吗？

Answer 1

终于破解了矢量化解决方案！这是一个有趣的问题。问题是我们必须标记从列表的相应数组元素中获取的每对值。然后，我们应该根据它们在其他对中的唯一性来标记每个这样的对。因此，我们可以使用np.unique 滥用其所有可选参数，最后做一些额外的工作来保持最终输出的顺序。这里的实现基本上分三个阶段完成 -

# Stack as a 2D array with each pair from values as a column each.
# Convert to linear index equivalent considering each column as indexing tuple
arr = np.vstack(values)
idx = np.ravel_multi_index(arr,arr.max(1)+1)

# Do the heavy work with np.unique to give us :
#   1. Starting indices of unique elems, 
#   2. Srray that has unique IDs for each element in idx, and 
#   3. Group ID counts
_,unq_start_idx,unqID,count = np.unique(idx,return_index=True, \
                                        return_inverse=True,return_counts=True)

# Best part happens here : Use mask to ignore the repeated elems and re-tag 
# each unqID using argsort() of masked elements from idx
mask = ~np.in1d(unqID,np.where(count>1)[0])
mask[unq_start_idx] = 1
out = idx[mask].argsort()[unqID]

运行时测试

让我们将提议的矢量化方法与原始代码进行比较。由于建议的代码只为我们提供组 ID，因此为了公平的基准测试，让我们从原始代码中删除不用于给我们的部分。所以，这里是函数定义 -

def groupify(values):  # Original code
    group = np.zeros((len(values[0]),), dtype=np.int64) - 1
    next_hash = 0
    matching = np.ones((len(values[0]),), dtype=bool)
    while any(group == -1):

        matching[:] = (group == -1)
        first_ungrouped_idx = np.where(matching)[0][0]

        for curr_id, value_array in enumerate(values):
            needed_value = value_array[first_ungrouped_idx]
            matching[matching] = value_array[matching] == needed_value
        # Assign all of the found elements to a new group
        group[matching] = next_hash
        next_hash += 1
    return group

def groupify_vectorized(values):  # Proposed code
    arr = np.vstack(values)
    idx = np.ravel_multi_index(arr,arr.max(1)+1)
    _,unq_start_idx,unqID,count = np.unique(idx,return_index=True, \
                                        return_inverse=True,return_counts=True)    
    mask = ~np.in1d(unqID,np.where(count>1)[0])
    mask[unq_start_idx] = 1
    return idx[mask].argsort()[unqID]

具有大型数组的列表上的运行时结果 -

In [345]: # Input list with random elements
     ...: values = [item for item in np.random.randint(10,40,(10,10000))]

In [346]: np.allclose(groupify(values),groupify_vectorized(values))
Out[346]: True

In [347]: %timeit groupify(values)
1 loops, best of 3: 4.02 s per loop

In [348]: %timeit groupify_vectorized(values)
100 loops, best of 3: 3.74 ms per loop

Answer 2

这应该可以工作，并且应该更快，因为我们正在使用broadcasting 和 numpy 固有的快速布尔比较：

import numpy as np

# Test values:
values = [
    np.array([10, 11, 10, 11, 10, 11, 10]),
    np.array([21, 21, 22, 22, 21, 22, 23]),
    ]
# Expected outcome: np.array([0, 1, 2, 3, 0, 3, 4])

# for every value in values, check where duplicate values occur
same_mask = [val[:,np.newaxis] == val[np.newaxis,:] for val in values]

# get the conjunction of all those tests
conjunction = np.logical_and.reduce(same_mask)

# ignore the diagonal
conjunction[np.diag_indices_from(conjunction)] = False

# initialize the labelled array with nans (used as flag)
labelled = np.empty(values[0].shape)
labelled.fill(np.nan)

# keep track of labelled value
val = 0
for k, row in enumerate(conjunction):
    if np.isnan(labelled[k]):  # this element has not been labelled yet
        labelled[k] = val      # so label it
        labelled[row] = val    # and label every element satisfying the test
        val += 1

print(labelled)
# outputs [ 0.  1.  2.  3.  0.  3.  4.]

在处理两个数组时，它比您的版本快大约 1.5 倍，但我怀疑对于更多数组，加速应该会更好。

Answer 3

numpy_indexed 包（免责声明：我是它的作者）包含 numpy 数组集操作的通用变体，可用于以优雅高效（矢量化）的方式解决您的问题：

import numpy_indexed as npi
unique_values, labels = npi.unique(tuple(values), return_inverse=True)

上述方法适用于任意类型组合，但如果 values 是许多具有相同 dtype 的数组的列表，则以下方法会更有效：

unique_values, labels = npi.unique(np.asarray(values), axis=1, return_inverse=True)

Answer 4

如果我理解正确，您正在尝试根据列对值进行哈希处理。最好将列自己转换为任意值，然后从中找到哈希值。

所以你实际上想要散列list(np.array(values).T)。

此功能已内置在 Pandas 中。你不需要写它。唯一的问题是它需要一个值列表，其中没有进一步的列表。在这种情况下，您只需将内部列表转换为 string map(str, list(np.array(values).T)) 并分解它！

>>> import pandas as pd
>>> pd.factorize(map(str, list(np.array(values).T)))
(array([0, 1, 2, 3, 0, 3, 4]),
 array(['[10 21]', '[11 21]', '[10 22]', '[11 22]', '[10 23]'], dtype=object))

我已将您的数组列表转换为数组，然后转换为字符串...