我用于切片 numpy 数组的代码(通过精美的索引)非常慢。它目前是程序中的一个瓶颈。
a.shape
(3218, 6)
ts = time.time(); a[rows][:, cols]; te = time.time(); print('%.8f' % (te-ts));
0.00200009
什么是正确的numpy调用来获取由矩阵a的行'rows'和列'col'的子集组成的数组?(其实我需要这个结果的转置)
让我尝试总结 Jaime 和 TheodrosZelleke 的出色答案,并加入一些评论。
a[rows][:,cols]
意味着两个花哨的索引操作,所以一个中间副本a[rows]
被创建和丢弃。方便且可读,但效率不高。此外请注意,[:,cols]
通常会从 C-cont 生成 Fortran 连续副本。来源。a[rows.reshape(-1,1),cols]
是一个单一的高级索引表达式,它基于rows.reshape(-1,1)
和cols
被广播到预期结果的形状。一个常见的经验是,扁平化数组中的索引可能比花式索引更有效,因此另一种方法是
indx = rows.reshape(-1,1)*a.shape[1] + cols
a.take(indx)
或者
a.take(indx.flat).reshape(rows.size,cols.size)
效率将取决于内存访问模式以及起始数组是 C-countinous 还是 Fortran 连续的,因此需要进行实验。
仅在真正需要时才使用花哨的索引:基本切片 a[rstart:rstop:rstep, cstart:cstop:cstep]
返回一个视图(尽管不是连续的)并且应该更快!
令我惊讶的是,这种计算第一个线性一维索引的长表达式比问题中提出的连续数组索引快50%以上:
(a.ravel()[(
cols + (rows * a.shape[1]).reshape((-1,1))
).ravel()]).reshape(rows.size, cols.size)
更新: OP 更新了初始数组形状的描述。使用更新后的大小,加速比现在超过99%:
In [93]: a = np.random.randn(3218, 1415)
In [94]: rows = np.random.randint(a.shape[0], size=2000)
In [95]: cols = np.random.randint(a.shape[1], size=6)
In [96]: timeit a[rows][:, cols]
10 loops, best of 3: 186 ms per loop
In [97]: timeit (a.ravel()[(cols + (rows * a.shape[1]).reshape((-1,1))).ravel()]).reshape(rows.size, cols.size)
1000 loops, best of 3: 1.56 ms per loop
初步答案: 以下是成绩单:
In [79]: a = np.random.randn(3218, 6)
In [80]: a.shape
Out[80]: (3218, 6)
In [81]: rows = np.random.randint(a.shape[0], size=2000)
In [82]: cols = np.array([1,3,4,5])
时间方法一:
In [83]: timeit a[rows][:, cols]
1000 loops, best of 3: 1.26 ms per loop
时间方法二:
In [84]: timeit (a.ravel()[(cols + (rows * a.shape[1]).reshape((-1,1))).ravel()]).reshape(rows.size, cols.size)
1000 loops, best of 3: 568 us per loop
检查结果是否实际上相同:
In [85]: result1 = a[rows][:, cols]
In [86]: result2 = (a.ravel()[(cols + (rows * a.shape[1]).reshape((-1,1))).ravel()]).reshape(rows.size, cols.size)
In [87]: np.sum(result1 - result2)
Out[87]: 0.0
如果您使用精美的索引和广播进行切片,则可以加快速度:
from __future__ import division
import numpy as np
def slice_1(a, rs, cs) :
return a[rs][:, cs]
def slice_2(a, rs, cs) :
return a[rs[:, None], cs]
>>> rows, cols = 3218, 6
>>> rs = np.unique(np.random.randint(0, rows, size=(rows//2,)))
>>> cs = np.unique(np.random.randint(0, cols, size=(cols//2,)))
>>> a = np.random.rand(rows, cols)
>>> import timeit
>>> print timeit.timeit('slice_1(a, rs, cs)',
'from __main__ import slice_1, a, rs, cs',
number=1000)
0.24083110865
>>> print timeit.timeit('slice_2(a, rs, cs)',
'from __main__ import slice_2, a, rs, cs',
number=1000)
0.206566124519
如果你从百分比的角度来考虑,做某事快 15% 总是好的,但在我的系统中,对于你的数组的大小,这需要 40 us 来进行切片,而且很难相信一个操作需要240我们将是你的瓶颈。
使用np.ix_
您可以与 ravel/reshape 相似的速度,但使用更清晰的代码:
a = np.random.randn(3218, 1415)
rows = np.random.randint(a.shape[0], size=2000)
cols = np.random.randint(a.shape[1], size=6)
a = np.random.randn(3218, 1415)
rows = np.random.randint(a.shape[0], size=2000)
cols = np.random.randint(a.shape[1], size=6)
%timeit (a.ravel()[(cols + (rows * a.shape[1]).reshape((-1,1))).ravel()]).reshape(rows.size, cols.size)
#101 µs ± 2.36 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)
%timeit ix_ = np.ix_(rows, cols); a[ix_]
#135 µs ± 7.47 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)
ix_ = np.ix_(rows, cols)
result1 = a[ix_]
result2 = (a.ravel()[(cols + (rows * a.shape[1]).reshape((-1,1))).ravel()]).reshape(rows.size, cols.size)
np.sum(result1 - result2)
0.0