【问题标题】：What is a faster way to get the location of unique rows in numpy什么是获取numpy中唯一行位置的更快方法
【发布时间】：2017-11-29 17:08:17
【问题描述】：

我有一个唯一行列表和另一个更大的数据数组（在示例中称为 test_rows）。我想知道是否有更快的方法来获取数据中每个唯一行的位置。我能想到的最快方法是……

import numpy


uniq_rows = numpy.array([[0, 1, 0],
                         [1, 1, 0],
                         [1, 1, 1],
                         [0, 1, 1]])

test_rows = numpy.array([[0, 1, 1],
                         [0, 1, 0],
                         [0, 0, 0],
                         [1, 1, 0],
                         [0, 1, 0],
                         [0, 1, 1],
                         [0, 1, 1],
                         [1, 1, 1],
                         [1, 1, 0],
                         [1, 1, 1],
                         [0, 1, 0],
                         [0, 0, 0],
                         [1, 1, 0]])

# this gives me the indexes of each group of unique rows
for row in uniq_rows.tolist():
    print row, numpy.where((test_rows == row).all(axis=1))[0]

这打印...

[0, 1, 0] [ 1  4 10]
[1, 1, 0] [ 3  8 12]
[1, 1, 1] [7 9]
[0, 1, 1] [0 5 6]

有没有更好或更 numpythonic（不确定该词是否存在）的方法来做到这一点？我正在寻找一个 numpy 组函数，但找不到它。基本上对于任何传入的数据集，我都需要最快的方法来获取该数据集中每个唯一行的位置。传入的数据集并不总是具有每个唯一的行或相同的数字。

编辑：这只是一个简单的例子。在我的应用程序中，数字不仅仅是零和一，它们可以是 0 到 32000 之间的任何值。uniq 行的大小可以在 4 到 128 行之间，而 test_rows 的大小可以达到数十万。

【问题讨论】：

对“numpythonic”投了赞成票。
数据总是只有0和1吗？
uniq_rows 和 test_rows 的典型大小（即行数和列数）是多少？
如果您尝试过发布的方法，是否有任何更新？
@Divakar 我离开了几天，所以我现在正在讨论解决方案。

标签： python numpy scipy

【解决方案1】：

方法#1

这是一种方法，虽然对于这样一个棘手的问题，但不确定“NumPythonic-ness”的水平 -

def get1Ds(a, b): # Get 1D views of each row from the two inputs
    # check that casting to void will create equal size elements
    assert a.shape[1:] == b.shape[1:]
    assert a.dtype == b.dtype

    # compute dtypes
    void_dt = np.dtype((np.void, a.dtype.itemsize * a.shape[1]))

    # convert to 1d void arrays
    a = np.ascontiguousarray(a)
    b = np.ascontiguousarray(b)
    a_void = a.reshape(a.shape[0], -1).view(void_dt).ravel()
    b_void = b.reshape(b.shape[0], -1).view(void_dt).ravel()
    return a_void, b_void

def matching_row_indices(uniq_rows, test_rows):
    A, B = get1Ds(uniq_rows, test_rows)
    validA_mask = np.in1d(A,B)

    sidx_A = A.argsort()
    validA_mask = validA_mask[sidx_A]    

    sidx = B.argsort()
    sortedB = B[sidx]
    split_idx = np.flatnonzero(sortedB[1:] != sortedB[:-1])+1
    all_split_indx = np.split(sidx, split_idx)

    match_mask = np.in1d(B,A)[sidx]
    valid_mask = np.logical_or.reduceat(match_mask, np.r_[0, split_idx])    
    locations = [e for i,e in enumerate(all_split_indx) if valid_mask[i]]

    return uniq_rows[sidx_A[validA_mask]], locations

改进范围（关于性能）：

np.split 可以替换为使用 slicing 进行拆分的 for 循环。
np.r_ 可以替换为 np.concatenate。

示例运行 -

In [331]: unq_rows, idx = matching_row_indices(uniq_rows, test_rows)

In [332]: unq_rows
Out[332]: 
array([[0, 1, 0],
       [0, 1, 1],
       [1, 1, 0],
       [1, 1, 1]])

In [333]: idx
Out[333]: [array([ 1,  4, 10]),array([0, 5, 6]),array([ 3,  8, 12]),array([7, 9])]

方法 #2

另一种克服前一个设置开销并从中使用get1Ds 的方法是 -

A, B = get1Ds(uniq_rows, test_rows)
idx_group = []
for row in A:
    idx_group.append(np.flatnonzero(B == row))

【讨论】：

v1.13 为unique 添加了一个axis 参数。你试过了吗？
@hpaulj 我仍然需要掌握最新版本。所以，不，我还没有。
方法 2 比我的解决方案快大约 6 倍。我还在玩它，但到目前为止它看起来很可靠。
@b10hazard 太棒了！从方法 #2 中删除了 print，并从列表中的每个组中收集索引，现在应该更容易计时。

【解决方案2】：

这样就可以了：

import numpy as np
uniq_rows = np.array([[0, 1, 0],
                         [1, 1, 0],
                         [1, 1, 1],
                         [0, 1, 1]])

test_rows = np.array([[0, 1, 1],
                         [0, 1, 0],
                         [0, 0, 0],
                         [1, 1, 0],
                         [0, 1, 0],
                         [0, 1, 1],
                         [0, 1, 1],
                         [1, 1, 1],
                         [1, 1, 0],
                         [1, 1, 1],
                         [0, 1, 0],
                         [0, 0, 0],
                         [1, 1, 0]])

indices=np.where(np.sum(np.abs(np.repeat(uniq_rows,len(test_rows),axis=0)-np.tile(test_rows,(len(uniq_rows),1))),axis=1)==0)[0]
loc=indices//len(test_rows)
indices=indices-loc*len(test_rows)
res=[[] for i in range(len(uniq_rows))]
for i in range(len(indices)):
    res[loc[i]].append(indices[i])
print(res)
[[1, 4, 10], [3, 8, 12], [7, 9], [0, 5, 6]]

这适用于所有情况，包括uniq_rows 中并非所有行都存在于test_rows 中的情况。但是，如果您事先知道它们都存在，则可以更换零件

res=[[] for i in range(len(uniq_rows))]
    for i in range(len(indices)):
        res[loc[i]].append(indices[i])

只有一行：

res=np.split(indices,np.where(np.diff(loc)>0)[0]+1)

从而完全避免循环。

【讨论】：

【解决方案3】：

麻木

从 numpy 1.13 版开始，您可以像 np.unique(test_rows, return_counts=True, return_index=True, axis=1) 一样使用 numpy.unique

熊猫

df = pd.DataFrame(test_rows)
uniq = pd.DataFrame(uniq_rows)

独特的

    0   1   2
0   0   1   0
1   1   1   0
2   1   1   1
3   0   1   1

或者您可以从传入的 DataFrame 自动生成唯一行

uniq_generated = df.drop_duplicates().reset_index(drop=True)

产量

    0   1   2
0   0   1   1
1   0   1   0
2   0   0   0
3   1   1   0
4   1   1   1

然后寻找它

d = dict()
for idx, row in uniq.iterrows():
    d[idx] = df.index[(df == row).all(axis=1)].values

这和你的where方法差不多

d

{0: array([ 1,  4, 10], dtype=int64),
 1: array([ 3,  8, 12], dtype=int64),
 2: array([7, 9], dtype=int64),
 3: array([0, 5, 6], dtype=int64)}

【讨论】：

不知道他们在 1.13 中更改了 np.unique。我会调查的。我不能使用 Pandas 解决方案，pandas 自身清理工作很糟糕，所以我不能使用它
pandas does a poor job cleaning up after itself 是什么意思？
我遇到了一个问题，我在项目中使用 pandas，但我无法让它释放所有内存。我在这里发布了我的问题：stackoverflow.com/questions/39100971/… 发布的解决方案对我没有用，所以我不再将 Pandas 用于生产代码。
当 numpy 做得很好时，没有理由使用 pandas。 numpy.unique 似乎是这里更好的解决方案。

【解决方案4】：

不是很“numpythonic”，但需要一些前期成本，我们可以使用键作为行的元组和索引列表来制作字典：

test_rowsdict = {}
for i,j in enumerate(test_rows):
    test_rowsdict.setdefault(tuple(j),[]).append(i)

test_rowsdict
{(0, 0, 0): [2, 11],
 (0, 1, 0): [1, 4, 10],
 (0, 1, 1): [0, 5, 6],
 (1, 1, 0): [3, 8, 12],
 (1, 1, 1): [7, 9]}

然后您可以根据您的 uniq_rows 进行过滤，并使用快速 dict 查找：test_rowsdict[tuple(row)]:

out = []
for i in uniq_rows:
    out.append((i, test_rowsdict.get(tuple(i),[])))

对于您的数据，仅查找需要 16us，构建和查找需要 66us，而您的 np.where 解决方案需要 95us。

【讨论】：

【解决方案5】：

使用 v1.13 中的 np.unique（从最新文档上的 source 链接下载，https://github.com/numpy/numpy/blob/master/numpy/lib/arraysetops.py#L112-L247）

In [157]: aset.unique(test_rows, axis=0,return_inverse=True,return_index=True)
Out[157]: 
(array([[0, 0, 0],
        [0, 1, 0],
        [0, 1, 1],
        [1, 1, 0],
        [1, 1, 1]]),
 array([2, 1, 0, 3, 7], dtype=int32),
 array([2, 1, 0, 3, 1, 2, 2, 4, 3, 4, 1, 0, 3], dtype=int32))

In [158]: a,b,c=_
In [159]: c
Out[159]: array([2, 1, 0, 3, 1, 2, 2, 4, 3, 4, 1, 0, 3], dtype=int32)
In [164]: from collections import defaultdict
In [165]: dd = defaultdict(list)
In [166]: for i,v in enumerate(c):
     ...:     dd[v].append(i)
     ...:     
In [167]: dd
Out[167]: 
defaultdict(list,
            {0: [2, 11],
             1: [1, 4, 10],
             2: [0, 5, 6],
             3: [3, 8, 12],
             4: [7, 9]})

或使用唯一行（作为可散列元组）索引字典：

In [170]: dd = defaultdict(list)
In [171]: for i,v in enumerate(c):
     ...:     dd[tuple(a[v])].append(i)
     ...:     
In [172]: dd
Out[172]: 
defaultdict(list,
            {(0, 0, 0): [2, 11],
             (0, 1, 0): [1, 4, 10],
             (0, 1, 1): [0, 5, 6],
             (1, 1, 0): [3, 8, 12],
             (1, 1, 1): [7, 9]})

【讨论】：

我试过了，但我的 timeit 测试显示它比我的解决方案慢。
@b10hazard，如果您将计时结果添加到问题中，我会很感兴趣。

【解决方案6】：

numpy_indexed 包（免责声明：我是它的作者）旨在以一种优雅高效的方式解决此类问题：

import numpy_indexed as npi
indices = np.arange(len(test_rows))
unique_test_rows, index_groups = npi.group_by(test_rows, indices)

如果你不关心所有行的索引，而只关心 test_rows 中存在的那些，npi 也有很多简单的方法来解决这个问题； f.i：

subset_indices = npi.indices(unique_test_rows, unique_rows)

作为旁注；查看 npi 库中的示例可能很有用；根据我的经验，大多数时候人们会问这类问题，这些分组索引只是达到目的的一种手段，而不是计算的最终目标。很有可能使用 npi 中的功能可以更有效地达到最终目标，而无需显式计算这些索引。您愿意为您的问题提供更多背景信息吗？

编辑：如果您的数组确实这么大，并且总是由少量具有二进制值的列组成，那么用以下编码包装它们可能会进一步提高效率：

def encode(rows):
    return (rows * [[2**i for i in range(rows.shape[1])]]).sum(axis=1, dtype=np.uint8)

【讨论】：

【解决方案7】：

这里有很多解决方案，但我要添加一个带有香草 numpy 的解决方案。在大多数情况下，numpy 会比列表解析和字典更快，尽管如果使用大型数组，数组广播可能会导致内存成为问题。

np.where((uniq_rows[:, None, :] == test_rows).all(2))

非常简单，嗯？这将返回唯一行索引的元组和相应的测试行。

 (array([0, 0, 0, 1, 1, 1, 2, 2, 3, 3, 3]),
  array([ 1,  4, 10,  3,  8, 12,  7,  9,  0,  5,  6]))

它是如何工作的：

(uniq_rows[:, None, :] == test_rows)

使用数组广播将test_rows 的每个元素与uniq_rows 中的每一行进行比较。这会产生一个 4x13x3 数组。 all 用于确定哪些行相等（所有比较都返回真）。最后，where 返回这些行的索引。

【讨论】：

这是最好的答案，谢谢，够简单，还搞定了两个索引系统，很赞！！！