在数组的两个不同部分计算重复整数的最快方法

Question

在这段 Python 代码中，

fun遍历数组arr并为每个部分对计算两个数组部分中相同整数的数量。（它模拟一个矩阵。）这n*(n-1)/2*m进行了总体比较，时间复杂度为O(n^2)。

是否有编程解决方案或重新构建此问题的方法可以产生等效的结果但降低了时间复杂度？

# n > 500000, 0 < i < n, m = 100

# dim(arr) = n*m, 0 < arr[x] < 4294967311

arr = mp.RawArray(ctypes.c_uint, n*m)

def fun(i):

    for j in range(i-1,0,-1):

        count = 0

        for k in range(0,m):

            count += (arr[i*m+k] == arr[j*m+k])

        if count/m > 0.7:

            return (i,j)

    return ()

arr 是一个共享内存阵列，因此为了简单和性能原因最好保持只读。

arr实现为来自 的一维原始数组multiprocessing。这样做的原因是，根据我的测试，它具有迄今为止最快的性能。numpy例如，使用2D 数组，如下所示：

arr = np.ctypeslib.as_array(mp.RawArray(ctypes.c_uint, n*m)).reshape(n,m)

将提供矢量化功能，但将总运行时间增加一个数量级 - 250 秒与 n = 1500 的 30 秒，即733%。

Answer 1

因此，在摆弄更多之后，我能够在 NumPy 的向量化和 Numba 的 JIT 编译器的帮助下大大减少运行时间。回到原来的代码：arr = mp.RawArray(ctypes.c_uint, n*m)def fun(i):&nbsp; &nbsp; for j in range(i-1,0,-1):&nbsp; &nbsp; &nbsp; &nbsp; count = 0&nbsp; &nbsp; &nbsp; &nbsp; for k in range(0,m):&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; count += (arr[i*m+k] == arr[j*m+k])&nbsp; &nbsp; &nbsp; &nbsp; if count/m > 0.7:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return (i,j)return ()我们可以省略最底层的return陈述，也可以count完全摒弃使用的想法，留给我们：def fun(i):&nbsp; &nbsp; for j in range(i-1,0,-1):&nbsp; &nbsp; &nbsp; &nbsp; if sum(arr[i*m+k] == arr[j*m+k] for k in range(m)) > 0.7*m:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return (i,j)然后，我们将数组更改arr为 NumPy 格式：np_arr = np.frombuffer(arr,dtype='int32').reshape(m,n)这里需要注意的重要一点是，我们不使用 NumPy 数组作为从多个进程写入的共享内存数组，从而避免了开销陷阱。最后，我们应用 Numba 的装饰器并sum以向量形式重写该函数，使其与新数组一起工作：import numba as [e-mail](fastmath=True,parallel=True)def fun(i):&nbsp; &nbsp; for j in range(i-1, 0, -1):&nbsp; &nbsp; &nbsp; &nbsp; if np.sum(np_arr[i] == np_arr[j]) > 0.7*m:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return (i,j)这将运行时间减少到7.9s，这对我来说绝对是一个胜利。

Answer 2

由于您根本无法更改数组特征，因此我认为您坚持使用O(n^2)。 numpy将获得一些矢量化，但会更改共享数组的其他人的访问权限。从最里面的操作开始：&nbsp; &nbsp; for k in range(0,m):&nbsp; &nbsp; &nbsp; &nbsp; count += (arr[i][k] == arr[j][k])将此更改为单行分配：&nbsp; &nbsp; count = sum(arr[i][k] == arr[j][k] for k in range(m))现在，如果这确实是一个数组，而不是一个列表列表，请使用数组包的向量化来简化循环，一次一个：&nbsp; &nbsp; count = sum(arr[i] == arr[j])&nbsp; &nbsp;# results in a vector of counts您现在可以返回j索引 where count[j] / m > 0.7。请注意，没有必要i为每一个都返回：它在函数内是常量，并且调用程序已经具有该值。您的数组包可能有一对可以返回这些索引的矢量化索引操作。如果您正在使用numpy，这些很容易在本网站上查找。