获取两个二维 numpy 数组的相交行

问题描述：

我想获取两个 2D numpy 数组中相交（公共）的行。例如，如果将以下数组作为输入传递：

array([[1, 4],
       [2, 5],
       [3, 6]])

array([[1, 4],
       [3, 6],
       [7, 8]])

输出应为：

array([[1, 4],
       [3, 6])

我知道如何使用循环来实现这一点。我正在寻找一种 Pythonic/Numpy 方法来做到这一点。

解决方案 1：

对于短数组，使用集合可能是最清晰、最易读的方法。

另一种方法是使用numpy.intersect1d。不过，你必须诱使它将行视为单个值……这会让事情变得不太可读……

import numpy as np

A = np.array([[1,4],[2,5],[3,6]])
B = np.array([[1,4],[3,6],[7,8]])

nrows, ncols = A.shape
dtype={'names':['f{}'.format(i) for i in range(ncols)],
       'formats':ncols * [A.dtype]}

C = np.intersect1d(A.view(dtype), B.view(dtype))

# This last bit is optional if you're okay with "C" being a structured array...
C = C.view(A.dtype).reshape(-1, ncols)

对于大型数组，这应该比使用集合快得多。

解决方案 2：

你可以使用 Python 的集合：

>>> import numpy as np
>>> A = np.array([[1,4],[2,5],[3,6]])
>>> B = np.array([[1,4],[3,6],[7,8]])
>>> aset = set([tuple(x) for x in A])
>>> bset = set([tuple(x) for x in B])
>>> np.array([x for x in aset & bset])
array([[1, 4],
       [3, 6]])

正如 Rob Cowie 指出的那样，这可以更简洁地完成，因为

np.array([x for x in set(tuple(x) for x in A) & set(tuple(x) for x in B)])

可能有一种方法可以做到这一点，而不需要从数组到元组的来回切换，但我现在还没想起来。

解决方案 3：

我不明白为什么没有建议使用纯 numpy 方法来做到这一点。所以我找到了一个使用 numpy 广播的方法。基本思想是通过轴交换将其中一个数组转换为 3d。让我们构造 2 个数组：

a=np.random.randint(10, size=(5, 3))
b=np.zeros_like(a)
b[:4,:]=a[np.random.randint(a.shape[0], size=4), :]

我跑了一下，结果显示：

a=array([[5, 6, 3],
   [8, 1, 0],
   [2, 1, 4],
   [8, 0, 6],
   [6, 7, 6]])
b=array([[2, 1, 4],
   [2, 1, 4],
   [6, 7, 6],
   [5, 6, 3],
   [0, 0, 0]])

步骤如下（数组可以互换）：

#a is nxm and b is kxm
c = np.swapaxes(a[:,:,None],1,2)==b #transform a to nx1xm
# c has nxkxm dimensions due to comparison broadcast
# each nxixj slice holds comparison matrix between a[j,:] and b[i,:]
# Decrease dimension to nxk with product:
c = np.prod(c,axis=2)
#To get around duplicates://
# Calculate cumulative sum in k-th dimension
c= c*np.cumsum(c,axis=0)
# compare with 1, so that to get only one 'True' statement by row
c=c==1
#//
# sum in k-th dimension, so that a nx1 vector is produced
c=np.sum(c,axis=1).astype(bool)
# The intersection between a and b is a[c]
result=a[c]

在一个有两行代码的函数中，用于减少内存使用量（如果有错请纠正我）：

def array_row_intersection(a,b):
   tmp=np.prod(np.swapaxes(a[:,:,None],1,2)==b,axis=2)
   return a[np.sum(np.cumsum(tmp,axis=0)*tmp==1,axis=1).astype(bool)]

这给出了我的示例的结果：

result=array([[5, 6, 3],
       [2, 1, 4],
       [6, 7, 6]])

这比集合解决方案更快，因为它只使用简单的 numpy 操作，同时不断减少维度，非常适合两个大矩阵。我想我的评论可能犯了错误，因为我是通过实验和直觉得到答案的。可以通过转置数组或稍微改变步骤来找到列交集的等价物。此外，如果需要重复项，则必须跳过“//”内的步骤。可以编辑该函数以仅返回索引的布尔数组，这对我来说很方便，同时尝试使用相同的向量获取不同的数组索引。投票答案和我的答案的基准（每个维度中的元素数量对选择什么有影响）：

代码：

def voted_answer(A,B):
    nrows, ncols = A.shape
    dtype={'names':['f{}'.format(i) for i in range(ncols)],
           'formats':ncols * [A.dtype]}
    C = np.intersect1d(A.view(dtype), B.view(dtype))
    return C.view(A.dtype).reshape(-1, ncols)

a_small=np.random.randint(10, size=(10, 10))
b_small=np.zeros_like(a_small)
b_small=a_small[np.random.randint(a_small.shape[0],size=[a_small.shape[0]]),:]
a_big_row=np.random.randint(10, size=(10, 1000))
b_big_row=a_big_row[np.random.randint(a_big_row.shape[0],size=[a_big_row.shape[0]]),:]
a_big_col=np.random.randint(10, size=(1000, 10))
b_big_col=a_big_col[np.random.randint(a_big_col.shape[0],size=[a_big_col.shape[0]]),:]
a_big_all=np.random.randint(10, size=(100,100))
b_big_all=a_big_all[np.random.randint(a_big_all.shape[0],size=[a_big_all.shape[0]]),:]



print 'Small arrays:'
print '     Voted answer:',timeit.timeit(lambda:voted_answer(a_small,b_small),number=100)/100
print '     Proposed answer:',timeit.timeit(lambda:array_row_intersection(a_small,b_small),number=100)/100
print 'Big column arrays:'
print '     Voted answer:',timeit.timeit(lambda:voted_answer(a_big_col,b_big_col),number=100)/100
print '     Proposed answer:',timeit.timeit(lambda:array_row_intersection(a_big_col,b_big_col),number=100)/100
print 'Big row arrays:'
print '     Voted answer:',timeit.timeit(lambda:voted_answer(a_big_row,b_big_row),number=100)/100
print '     Proposed answer:',timeit.timeit(lambda:array_row_intersection(a_big_row,b_big_row),number=100)/100
print 'Big arrays:'
print '     Voted answer:',timeit.timeit(lambda:voted_answer(a_big_all,b_big_all),number=100)/100
print '     Proposed answer:',timeit.timeit(lambda:array_row_intersection(a_big_all,b_big_all),number=100)/100

结果：

Small arrays:
     Voted answer: 7.47108459473e-05
     Proposed answer: 2.47001647949e-05
Big column arrays:
     Voted answer: 0.00198730945587
     Proposed answer: 0.0560171294212
Big row arrays:
     Voted answer: 0.00500325918198
     Proposed answer: 0.000308241844177
Big arrays:
     Voted answer: 0.000864889621735
     Proposed answer: 0.00257176160812

下面的结论是，如果您必须比较 2 个大的 2d 点数组，则使用投票答案。如果您在所有维度上都有大矩阵，投票答案无疑是最好的答案。所以，这取决于您每次选择什么。

解决方案 4：

Numpy broadcasting

我们可以使用广播创建一个布尔掩码，然后可以使用它来过滤数组A中也存在的行B

A = np.array([[1,4],[2,5],[3,6]])
B = np.array([[1,4],[3,6],[7,8]])

m = (A[:, None] == B).all(-1).any(1)

>>> A[m]

array([[1, 4],
       [3, 6]])

解决方案 5：

实现此目的的另一种方法是使用结构化数组：

>>> a = np.array([[3, 1, 2], [5, 8, 9], [7, 4, 3]])
>>> b = np.array([[2, 3, 0], [3, 1, 2], [7, 4, 3]])
>>> av = a.view([('', a.dtype)] * a.shape[1]).ravel()
>>> bv = b.view([('', b.dtype)] * b.shape[1]).ravel()
>>> np.intersect1d(av, bv).view(a.dtype).reshape(-1, a.shape[1])
array([[3, 1, 2],
       [7, 4, 3]])

为了清楚起见，结构化视图如下所示：

>>> a.view([('', a.dtype)] * a.shape[1])
array([[(3, 1, 2)],
       [(5, 8, 9)],
       [(7, 4, 3)]],
       dtype=[('f0', '<i8'), ('f1', '<i8'), ('f2', '<i8')])

解决方案 6：

np.array(set(map(tuple, b)).difference(set(map(tuple, a))))

这也可以

解决方案 7：

A = np.array([[1,4],[2,5],[3,6]])
B = np.array([[1,4],[3,6],[7,8]])

def matching_rows(A,B):
  matches=[i for i in range(B.shape[0]) if np.any(np.all(A==B[i],axis=1))]
  if len(matches)==0:
    return B[matches]
  return np.unique(B[matches],axis=0)

>>> matching_rows(A,B)
array([[1, 4],
       [3, 6]])

当然，这假设各行的长度都相同。

解决方案 8：

无需索引
访问https://gist.github.com/RashidLadj/971c7235ce796836853fcf55b4876f3c

def intersect2D(Array_A, Array_B):
"""
Find row intersection between 2D numpy arrays, a and b.
"""

# ''' Using Tuple ''' #
intersectionList = list(set([tuple(x) for x in Array_A for y in Array_B  if(tuple(x) == tuple(y))]))
print ("intersectionList = 
",intersectionList)

# ''' Using Numpy function "array_equal" ''' #
""" This method is valid for an ndarray """
intersectionList = list(set([tuple(x) for x in Array_A for y in Array_B  if(np.array_equal(x, y))]))
print ("intersectionList = 
",intersectionList)

# ''' Using set and bitwise and '''
intersectionList = [list(y) for y in (set([tuple(x) for x in Array_A]) & set([tuple(x) for x in Array_B]))]
print ("intersectionList = 
",intersectionList)

return intersectionList

使用索引
访问https://gist.github.com/RashidLadj/bac71f3d3380064de2f9abe0ae43c19e

def intersect2D(Array_A, Array_B):
  """
  Find row intersection between 2D numpy arrays, a and b.
  Returns another numpy array with shared rows and index of items in A & B arrays
  """
  # [[IDX], [IDY], [value]] where Equal
  # ''' Using Tuple ''' #
  IndexEqual = np.asarray([(i, j, x) for i,x in enumerate(Array_A) for j, y in enumerate (Array_B)  if(tuple(x) == tuple(y))]).T
  
  # ''' Using Numpy array_equal ''' #
  IndexEqual = np.asarray([(i, j, x) for i,x in enumerate(Array_A) for j, y in enumerate (Array_B)  if(np.array_equal(x, y))]).T
  
  idx, idy, intersectionList = (IndexEqual[0], IndexEqual[1], IndexEqual[2]) if len(IndexEqual) != 0 else ([], [], [])

  return intersectionList, idx, idy

解决方案 9：

当数组非常大时，上述所有方法都很慢。我找到了一种使用 numpy 的方法。由于 numpy 仅为 1D 数组提供np.in1d ，我们可以使用轴 1 中的康托配对将 2D 数组编码为 1D 数组。然后就可以使用 numpy 的函数了。

def cantor_pairing(a, b):
    return (a + b) * (a + b + 1) / 2 + a

def intersecting_indices(a, b):

    pair_a = cantor_pairing(cantor_pairing(a[:,0], a[:, 1]), a[:, 2])
    pair_b = cantor_pairing(cantor_pairing(b[:,0], b[:, 1]), b[:, 2])

    boolean_array = np.in1d(pair_a, pair_b)
    intersected_indices = np.where(bool_array==True)[0]

    return intersected_indices

对于非常大的数组，这是最快的方法。需要记住的一件重要事情是，康托配对只能对非负整数进行。因此，如果数组中有负整数，请在使用之前将它们全部变为正数（通过添加最小值）。

解决方案 10：

import numpy as np

A=np.array([[1, 4],
       [2, 5],
       [3, 6]])

B=np.array([[1, 4],
       [3, 6],
       [7, 8]])

intersetingRows=[(B==irow).all(axis=1).any() for irow in A]
print(A[intersetingRows])

解决方案 11：

您可以使用 numpy.intersect 函数以及设置为 0 的 axis 参数来实现此目的，以查找两个 2D numpy 数组之间的公共行。以下是示例：

      import numpy as np

     # Define the input arrays
array1 = np.array([[1, 4],
                   [2, 5],
                   [3, 6]])

array2 = np.array([[1, 4],
                   [3, 6],
                   [7, 8]])

# Find the intersecting rows
result = np.intersect1d(array1, array2, assume_unique=True)

# Reshape the result back into 2D array
result = result.reshape(-1, array1.shape[1])

print(result)

输出：

[[1 4]
 [3 6]]

本例中使用np.intersect1d来找出array1和array2共同的元素。在已知输入数组唯一的情况下，使用assume_unique=True来进行性能优化。

然后，将结果重塑为二维数组以获得最终输出。

这种方法提供了一种更符合 Pythonic/Numpy 的方式来查找两个 2D numpy 数组之间的相交行，而无需使用显式循环。