python - 如何在 Python 中并行化列表元素

Question

我有一个很大的字典列表。我想从这个列表中计算一个二维矩阵，其中每个元素都是列表中第 i 个元素和列表中第 j 个元素的函数。代码是这样的:

    matrix=np.array([])
    biglist=self.some_method()
    matrix.resize(len(biglist),len(biglist))

    for i in range(len(biglist)):
        for j in range(i,len(biglist)):
            matrix[i,j]=self.__computeScore(biglist[i], biglist[j], i, j)[2]
            matrix[j,i]=matrix[i,j]

和 __computeScore现在的方法很简单:

def __computeScore(self, dictionary1, dictionary2, i, j):
    #value=in future some computation over dictionary1 and dictionary2
    value=1 #for now is so
    return (i,j,value)

即使计算分数的方法很简单，现在计算矩阵也需要一段时间。我想并行化矩阵计算。最好的方法是什么？到目前为止，我已经尝试使用 apply_async和 Pool.map来自 multiprocessing模块，但计算花费的时间比原始代码还要多。我试过类似的东西:

    pool = multiprocessing.Pool(processes=4)
    params=[]
    print "argument creation" #this takes a while so it's not convinient
    for i in range(len(biglist)):
        for j in range(i,len(biglist)):
            params.append((biglist[i],biglist[j],i,j))

    result=pool.map(self.__computeScore, params) #takes more time than the original code

我也尝试过类似的东西:

    def my_callback( result ):
        matrix[result[0],result[1]]=result[2]

    pool = multiprocessing.Pool(processes=4)
    for i in range(len(biglist)):
        rcopy=dict(biglist[i])
        for j in range(i,len(biglist)):
            ccopy=dict(biglist[j])
            pool.apply_async(self.__computeScore, args=(rcopy, ccopy, i, j), callback = my_callback)
    pool.close()
    pool.join()

但它比原始代码花费更多时间。我哪里错了？谢谢

Answer 1

Where am I wrong?

假设在 matrix[i,j]=self.__computeScore(...) 级别的多个进程之间并行化将为您带来显着的性能提升。

此外，假设您的代码的微小修改将导致显着加速。对于像您这样的“数学”问题，情况不一定如此。这通常需要您重新构建算法，涉及 更高效的数据结构 .

为什么让你失望

事实上，您一直观察到的是，生成进程和向它们传达任务带有 。巨大的与将事物保持在一个进程甚至线程中相比的开销。这种开销只有在 时才会得到返回。传达任务与 相比，花费的时间要少得多计算时间任务需要。您的基于多处理的方法将大部分时间花在进程间通信上， __computeScore(self, dictionary1, dictionary2, i, j)很可能只是让子进程感到无聊。一个反例:假设你给一个进程一个文件名，然后这个进程花费 15 秒从该文件读取数据并处理该数据。传输文件名只需几微秒。因此，在这种情况下，“定义”工作所花费的时间与执行实际工作所花费的时间相比可以忽略不计。这是多处理大放异彩的场景。它实际上非常简单，尤其是一旦了解了多处理在幕后的工作原理。但是，不幸的是，您在这里的误解是一个非常普遍的误解。

你的情况完全不同。您需要了解计算机的工作原理才能了解如何优化您的应用案例。首先，您应该防止在 RAM 中复制大量数据，尤其是在进程空间之间!你操作的大数据应该存储 一次内存，采用有效的数据结构(即，采用允许以有效方式执行所有后续操作的数据结构)。然后你需要意识到 Python 本身并不是一门快速的语言。解决方案不是使用更快的高级语言，而是尽可能将性能关键代码“外部化”为编译代码——你拥有的双重嵌套循环是一个强有力的指标。此循环应在已编译的例程中执行。您已经将问题视为线性代数问题这一事实强烈要求更多地使用线性代数包(您已经使用 numpy ，但不是因此)。我看到您需要循环，因为您的许多“基础”数据都在列表或字典中。在我看来，您需要找到一种方法来摆脱这些列表/字典，并根据 numpy 数据类型定义您拥有的所有数据。这样，您可能会找到一种简单的方法来使 O(N^2) 操作由机器代码执行，而不是由您的 Python 解释器执行。

结论——你应该如何处理这个

您不需要多个进程。您需要在一个进程(甚至线程)中使用正确的方法。你需要:

更高效的数据结构

用于计算的基于线性代数的算法

用于执行此算法的高度优化的数学库

如果您以正确的方式处理此问题，我可以向您保证，您的代码运行速度会快几个数量级。因为 numpy/scipy 在底层利用了基于矢量化的 CPU 功能(SIMD，“单指令，多数据”)，这将使您的数据操作以极快的速度运行。而且，如果您的矩阵足够大，以至于您的算法在理论上可以利用多个 CPU 内核:在 numpy 和 scipy 中甚至还支持 OpenMP，通过这些支持，某些矩阵操作例程可以自动将其工作分配到多个线程上。