python - 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仍然以某种方式引用旧的组件数组，或者另一种方法来实现数组之间的 Hadamard 乘积的速度 stacked在改变未堆叠的那些？如果不需要进行拆垛 (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] 列表是更大的 1D 数组列表的子列表。 这个更大的列表是 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

让我们把问题分解一下。您希望拥有对所有可变数组的引用列表，但您希望能够将它们作为一个 block 来执行操作。

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

替换你当前的循环

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 就会反射(reflect)所有更改。

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

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

 indices = np.arange(M)

 # inside the loop:

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

只要 M 不变并且您相信洗牌足够随机，您就不需要重新分配或重新排序 indices。