【问题标题】:Hybridizing Numpy vectorization and plain Python混合 Numpy 矢量化和普通 Python
【发布时间】:2023-03-28 18:25:01
【问题描述】:

我有一个二维ndarray,应扫描其中的行以检查是否有任何一个等于任何其他。

我的第一次尝试确实有效,但我觉得这不是最佳方式。一旦矩阵中的行数接近1000,就需要时间。

我的代码如下。 X 是前面提到的数组,Y 也是一个二维的ndarray

for i in range(X.shape[0]-1):
    for j in range(i+1,X.shape[0]):
        if (np.all( (X[i,:] == X[j,:] ), axis = 0 )):
            Y[j,:] = Y[i,:]
        #endif
    #enddo
#enddo

我知道嵌套循环很耗时,应该避免,但我找不到替代方法。列表理解在我看来并不合适,因为不需要保存项目。

该过程的核心是赋值操作Y[j,:] = Y[i,:],它依赖于索引,这将导致我排除类似列表理解的解决方案。

问题然后是:有没有更有效的方法来编写利用numpy 向量化的搜索?

【问题讨论】:

  • 输入数组X的典型形状是什么?
  • @Divakar X 被认为是一个合成数据集。理想情况下,我会使用 (1000,8) 的形状,甚至可能会有更多行

标签: python python-3.x numpy vectorization


【解决方案1】:

请参见以下示例: 作为一个例子,考虑一个 True 和 False 的一维向量,您想要计算序列中“假到真”转换的数量:

np.random.seed(444)
x = np.random.choice([False, True], size=100000)

使用 Python for 循环,一种方法是成对评估序列中每个元素的真值以及紧随其后的元素:

def count_transitions(x) -> int:
  count = 0
  for i, j in zip(x[:-1], x[1:]):
     if j and not i:
        count += 1
  return count

count_transitions(x)

在矢量化形式中,没有明确的 for 循环或对单个元素的直接引用:

np.count_nonzero(x[:-1] < x[1:])

这两个等效函数在性能方面如何比较?在这种特殊情况下,向量化的 NumPy 调用胜出大约 70 倍

https://realpython.com/numpy-array-programming/

【讨论】:

  • 谢谢你的回答,这是一个聪明的把戏。但这回答了 多少 行彼此相等的问题,但这里主要关注的是 哪些 相等?我的意思是:在报告的示例以及您链接的页面中,我找不到这样的矢量化黑客
【解决方案2】:

方法#1

我们可以利用行视图来获得成对匹配。然后,运行循环并将它们分配给Y。这个想法是在我们开始运行循环后最小化工作。考虑到可能有多个索引与其他索引匹配,因此很难提出纯矢量化方法。实现看起来像这样 -

# https://stackoverflow.com/a/44999009/ @Divakar
def view1D(a): # a is array
    a = np.ascontiguousarray(a)
    void_dt = np.dtype((np.void, a.dtype.itemsize * a.shape[1]))
    return a.view(void_dt).ravel()

# Get 1D view
a1D = view1D(a)

# Perform broadcasting to get outer equality match
mask = a1D[:,None]==a1D

# Get indices of pairwise matches
n = len(mask)
mask[np.tri(n, dtype=bool)] = 0
idx = np.argwhere(mask)

# Run loop to assign equal rows in Y
for (i,j) in zip(idx[:,0],idx[:,1]):
    Y[j] = Y[i]

备选方案#1:使用掩码直接赋值

所以,用mask,直接分配Y中的行,像这样-

for i,m in enumerate(mask):
    if m.any():
        Y[m] = Y[i]

如果有很多匹配项,这将很有帮助。

备选方案 #2:使用缩减蒙版

如果两行之间有多个公共行,我们可能希望减少这些行,以使所有这些行都与第一个出现的行相关联。因此,我们可以生成一个 reduced-mask 并使用它来代替之前的 mask -

mask0 = np.zeros_like(mask)
mask0[mask.argmax(0), np.arange(len(mask))] = 1
np.fill_diagonal(mask0,0)

然后,使用mask0 代替mask 并分配。


方法#2

另一种方法是从 1D 视图方法开始,并使用基于排序的方法来设置成对匹配索引,就像这样 -

sidx = a1D.argsort() # a1D from earlier approach
b = a1D[sidx]
m0 = b[:-1] == b[1:]
m1 = np.r_[False,m0,False]
idx = np.flatnonzero(m1[:-1]!=m1[1:]).reshape(-1,2)
for (i,j) in idx:
    row0,row1 = sidx[i],sidx[i+1:j+1]
    Y[row1] = Y[row0]

【讨论】:

  • 这是一个非常有趣的方法!它是否可能类似于带有偏移量的矩阵作为数组的表示,就像在 C 代码优化中常见的那样?
  • @MatteoZambra 不知道matrix-as-array-with-offset representation。你能详细说明一下吗?
  • 也许我误解了您答案的基本逻辑,但我看到了一些 C 代码优化示例,其中二维数组,例如 int mat[N][M] 实际上被视为一维数组 int mat[N + M],索引位置为mat[i*M + j] 而不是mat[i][j]。据我所知,它是为了加快 ip 获取过程的速度,即以线性方式保存二维数组,与 RAM 的线性特征一致。有 [this]("stackoverflow.com/a/14015582/9136498") 帖子和其中的引用进一步加深
  • @MatteoZambra Nah,这不会发生在这里。说row2 = [4,5,8]row1 = 2,然后使用Y[row1] = Y[row0],首先它将Y[row0] 广播到三个副本,然后一次性分配-Y[4] = Y[2]Y[5] = Y[2]Y[8] = Y[2]。在引擎盖下,它不需要制作Y[2] 的那些副本,所以它非常有效。希望这是有道理的。
【解决方案3】:

我正在使用手机,因此无法对此进行测试,但我认为它会起作用

mask = np.all(X[:, None] == X[None], axis=-1)
ind1, ind2 = np.nonzero(mask)
ind1, ind2 = ind1[ind1 < ind2], ind2[ind1 < ind2]
Y[ind2] = Y[ind1]

【讨论】:

  • 感谢您的回答。我没有得到axis = -1 子句。我在网上寻找,但找不到更多信息。那个轴是什么意思?
  • 这意味着np.all 检查数组的所有元素是否跨越last 轴,对于其他轴的每个索引。在这种情况下,它将与axis=2 相同。您还可以使用整数元组来检查多个轴。默认情况下,np.all 检查每个轴。
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 2020-01-16
  • 1970-01-01
  • 2018-03-17
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多