python - 在Python中迭代numpy数组的更快方法是什么

Question

我想知道是否有更好的方法来迭代 numpy 数组？我已经对嵌套迭代进行了计时，每个循环大约需要 40-50 秒，我想知道是否有更快的方法来做到这一点？我知道循环 numpy 数组并不理想，但我没有想法。我在 Stack Overflow 上查看了很多问题，但所有这些问题最终都让我更加困惑。

我尝试使用 tolist() 函数将 numpy 数组转换为列表，但是运行时间即使不是更差，也同样慢。

def euc_distance(array1, array2):
    return np.power(np.sum((array1 - array2)**2) , 0.5)

for i in range(N):
    for j,n in enumerate(data2.values): 
        distance = euc_distance(n, D[i]) 
        if distance < Dradius[i] and NormAttListTest[j] == "Attack":
            TP += 1

我的 euc_distance 函数传入数组形式(在我的例子中为 5 维)输入，以输出 1 维值。我的 data2.values 是我通过 pandas 框架访问 numpy 数组的方式，它是一个 [500 000, 5] 数据帧。

(请注意，NormAttListTest 是一个列表，其中每个单独的测试数据都标记有“攻击”和“正常”的分类数据)。

Answer 1

您的问题是您以错误的方式使用numpy，因为numpy都是关于矢量化计算的，例如MATLAB。考虑对您的代码进行以下修改。我用简单的 numpy 代码替换了 numpy 数组上的循环，该代码可以有效地利用二维数组的矢量化。结果，代码运行速度提高了 100 倍。

import functools
import numpy as np
import time

# decorator to measure running time
def measure_running_time(echo=True):
    def decorator(func):
        @functools.wraps(func)
        def wrapped(*args, **kwargs):
            t_1 = time.time()
            ans = func(*args, **kwargs)
            t_2 = time.time()
            if echo:
                print(f'{func.__name__}() running time is {t_2 - t_1:.2f} s')
            return ans
        return wrapped
    return decorator


def euc_distance(array1, array2):
    return np.power(np.sum((array1 - array2) ** 2), 0.5)

# original function
@measure_running_time()
def calculate_TP_1(N, data2, D, Dradius, NormAttListTest, TP=0):
    for i in range(N):
        for j, n in enumerate(data2):
            distance = euc_distance(n, D[i])
            if distance < Dradius[i] and NormAttListTest[j] == "Attack":
                TP += 1
    return TP

# new version
@measure_running_time()
def calculate_TP_2(N, data2, D, Dradius, NormAttListTest, TP=0):

    # this condition is the same for every i value
    NormAttListTest = np.array([val == 'Attack' for val in NormAttListTest])

    for i in range(N):

        # don't use loop over numpy arrays

        # compute distance for all the rows
        distance = np.sum((data2 - D[i]) ** 2, axis=1) ** .5
        # check conditions for all the row
        TP += np.sum((distance < Dradius[i]) & (NormAttListTest))

    return TP


if __name__ == '__main__':

    N = 10
    NN = 100_000
    D = np.random.randint(0, 10, (N, 5))
    Dradius = np.random.randint(0, 10, (N,))
    NormAttListTest = ['Attack'] * NN
    NormAttListTest[:NN // 2] = ['Defence'] * (NN // 2)
    data2 = np.random.randint(0, 10, (NN, 5))

    print(calculate_TP_1(N, data2, D, Dradius, NormAttListTest))

    print(calculate_TP_2(N, data2, D, Dradius, NormAttListTest))

输出:

calculate_TP_1() running time is 7.24 s
96476
calculate_TP_2() running time is 0.06 s
96476