python - 使用Python多处理池时系统内存不足？

Question

我正在尝试使用 Python 中的多处理模块并行化我的代码以查找相似度矩阵。当我使用带有 10 X 15 元素的小型 np.ndarray 时，它工作得很好。但是，当我将 np.ndarray 缩放到 3613 X 7040 元素时，系统内存不足。

下面是我的代码。

import multiprocessing 
from multiprocessing import Pool
## Importing Jacard_similarity_score
from sklearn.metrics import jaccard_similarity_score

# Function for finding the similarities between two np arrays
def similarityMetric(a,b):
    return (jaccard_similarity_score(a,b))

## Below functions are used for Parallelizing the scripts
 # auxiliary funciton to make it work
def product_helper1(args):
    return (similarityMetric(*args))

def parallel_product1(list_a, list_b):
    # spark given number of processes
    p = Pool(8) 
    # set each matching item into a tuple
    job_args = getArguments(list_a,list_b)    
    # map to pool
    results = p.map(product_helper1, job_args)
    p.close()
    p.join()
    return (results)

## getArguments function is used to get the combined list 
def getArguments(list_a,list_b):
    arguments = []
    for i in list_a:
        for j in list_b:
            item = (i,j)
            arguments.append(item)
    return (arguments)

现在，当我运行以下代码时，系统内存不足并挂起。我传递了两个 numpy.ndarrays testMatrix1 和 testMatrix2，其大小为 (3613, 7040)

resultantMatrix = parallel_product1(testMatrix1,testMatrix2)

我是在 Python 中使用这个模块的新手，并试图了解我哪里出错了。如有任何帮助，我们将不胜感激。

Answer 1

问题很可能只是组合爆炸。您试图预先实现主进程中的所有对，而不是实时生成它们，因此您需要存储大量内存。假设 ndarray 包含 double 值，这些值变为 Python float，则返回的 list 的内存使用情况getArguments 大致是每对 tuple 和两个 float 的成本，或者大约是:

3613 * 7040 * (sys.getsizeof((0., 0.)) + sys.getsizeof(0.) * 2)

在我的 64 位 Linux 系统上，这意味着在工作人员执行任何操作之前，Py3 上需要约 2.65 GB RAM，或者 Py2 上需要约 2.85 GB RAM。

如果您可以使用生成器以流式处理方式处理数据，因此参数会延迟生成并在不再需要时被丢弃，您可能会显着减少内存使用量:

import itertools

def parallel_product1(list_a, list_b):
    # spark given number of processes
    p = Pool(8) 
    # set each matching item into a tuple
    # Returns a generator that lazily produces the tuples
    job_args = itertools.product(list_a,list_b)    
    # map to pool
    results = p.map(product_helper1, job_args)
    p.close()
    p.join()
    return (results)

这仍然需要所有结果都适合内存；如果product_helper返回floats，那么在64位机器上resultlist的预期内存使用量仍然会大约 0.75 GB 左右，相当大；如果您可以以流方式处理结果，迭代 p.imap 的结果，甚至更好，p.imap_unordered (后者返回计算结果，而不是在命令生成器生成参数)并将它们写入磁盘或以其他方式确保它们快速在内存中释放将节省大量内存；下面的代码只是将它们打印出来，但以某种可重新摄取的格式将它们写入文件也是合理的。

def parallel_product1(list_a, list_b):
    # spark given number of processes
    p = Pool(8) 
    # set each matching item into a tuple
    # Returns a generator that lazily produces the tuples
    job_args = itertools.product(list_a,list_b)    
    # map to pool
    for result in p.imap_unordered(product_helper1, job_args):
        print(result)
    p.close()
    p.join()