Hadamard 在数组列表上的效率

Question

在一维数组的垂直堆栈上的 numpy，一个 Hadmard 乘积明显快于循环遍历一维数组列表并在每个数组上执行 Hadamard（逐元素）乘积（这是有道理的，我还是测试了它） .

我有一种情况，我需要在一组 numpy 数组和另一组数组之间执行 Hadamard 乘积：

stacked_arrays = np.vstack([1D-arrays...])
stacked_arrays *= np.power(factor, np.arange(1, num_arrays))

然而，我需要这个操作来改变列表中每个组件的一维数组，而且这个操作需要经常发生。我知道这听起来像一个奇怪的功能，但有没有办法做到这一点没有像这样的循环：

factors = factor ** np.arange(1, num_arrays)
for array, f in zip([1D..arrays], factors):
     array *= f

还是不将运算结果拆开？

也不能使用map，因为map 会创建 numpy 数组的副本：

result = map(lambda x, y: x * y, zip([1D..arrays], factors))

因为你不能用lambda 做*=，所以会返回一个 numpy 数组列表，而原始数组保持不变。

有没有办法让np.vstack 仍然以某种方式引用旧的组件数组，或者另一种方法来实现stacked 数组之间的 Hadamard 乘积的速度，同时改变未堆叠的数组？如果不需要取消堆叠 (np.split)，可以节省一些时间。

TimeIt 结果：

m = []
for _ in range(100):
    m.append(np.array([1, 2, 4, 5], dtype=np.float64))
factors = np.expand_dims(np.power(2, np.arange(100, dtype=np.float64)), axis=1)

def split_and_unstack():
    l = np.vstack(m)
    l *= factors
    result = np.split(l, 100)

def split_only():
    l = np.vstack(m)
    l *= factors

print(timeit.timeit(split_and_unstack, number=10000))
# 1.8569015570101328

print(timeit.timeit(split_only, number=10000))
# 0.9328480050317012

# makes sense that with unstacking it is about double the time

澄清： 上面提到的 [1D arrays] 列表是更大的一维数组列表的子列表。 这个更大的列表是collections.deque。而这个deque 需要 在提取子列表之前进行洗牌（即，这是随机梯度下降的体验重放缓冲区）。

缓冲pop和append速度：

times = int(1e4)
tgt = np.array([1, 2, 3, 4])

queue = collections.deque([tgt] * times, maxlen=times)
reg_list = [tgt] * times
numpy_list = np.array([tgt] * times)

def pop():
    queue.pop()
def pop_list():
    reg_list.pop()
def pop_np():
    global numpy_list
    numpy_list = numpy_list[1:]

print(timeit.timeit(pop, number=times))
# 0.0008135469979606569
print(timeit.timeit(pop_list, number=times))
# 0.000994370027910918
print(timeit.timeit(pop_np, number=times))
# 0.0016436030273325741

def push():
    queue.append(tgt)
def push_list():
    reg_list.append(tgt)
numpy_list = np.array([tgt] * 1)
def push_np():
    numpy_list[0] = tgt

print(timeit.timeit(push, number=times))
# 0.0008797429618425667
print(timeit.timeit(push_list, number=times))
# 0.00097957398975268
print(timeit.timeit(push_np, number=times))
# 0.003331452957354486

Answer 1

让我们分解问题。您希望拥有一个对所有可变数组的引用列表，但您希望能够以块的形式对它们执行操作。

我认为你的方法是倒退的。与其尝试将数组打包和解包到单独的缓冲区中，不如将视图维护到单个缓冲区中。

替换当前循环

m = [np.array([1, 2, 4, 5], dtype=np.float64) for _ in range(100)]

使用单个缓冲区，并查看每一行：

buf = np.vstack([np.array([1, 2, 4, 5], dtype=np.float64) for _ in range(100)])
m = list(buf)  # check that m[0].base is b

现在您有一个数组列表m，您可以单独修改每个数组。只要您将修改保留在原地并且不重新分配列表元素，所有更改都将直接出现在buf 中。同时，你可以对buf进行批量计算，只要你在原地做，m就会反映所有的变化。

实际上，您甚至可能不需要m。请注意list(buf) 如何为块的每一行创建视图。您可以轻松地直接索引到缓冲区中。例如，m[3][8] 通常将buf 写成buf[3, 8]，但您也可以使用buf 是一个序列这一事实并写成buf[3][8]。这效率较低，因为它每次都会创建一个新视图 (buf[3])，但不会太多。

要提取行的洗牌子集，您也可以使用序列索引。假设您的缓冲区保留了最后的M 行，您希望从中洗牌并提取N 行的子序列。您可以通过创建一个索引数组并一遍又一遍地改组这些索引来做到这一点：

 indices = np.arange(M)

 # inside the loop:

 np.random.shuffle(indices)
 chunk = buf[indices[:N]]
 # do your math on `chunk`

您不需要重新分配或重新排序 indices，只要 M 不改变并且您认为随机播放足够随机。