多处理的池方法的性能问题

Question

我在数据框中有一列，它由项目列表组成，我想根据 Fisher 精确测试计算该数据框(在本例中将列出(的行与所有其他行的相似性。为此，我想使用 python 多处理中的Pool，但它似乎与传统方法(即使用嵌套的 for 循环(花费的时间大致相同。有什么方法可以优化代码吗？

费舍尔试验

def fisher_test(a, b, c, d):
# do some stuff and return p value

使用嵌套 for 循环的计算：

%%time
import multiprocessing as mp
pool = mp.Pool(mp.cpu_count())

universeSize = 13000
# gq_result_df is a data frame
for i, row in gq_result_df.iterrows():
for j in range(i, gq_result_df.shape[0]):
if(i==j):
continue
pval = fisher_test(row["module_genes"], gq_result_df.loc[j,"module_genes"], universeSize)
# pval_matrix is a matrix in which we are storing the result
pval_matrix[i,j] = pval

使用池并行化内部循环：

%%time
universeSize = 13000
import multiprocessing as mp
pool = mp.Pool(mp.cpu_count())
for i, row in range(0, gq_result_df.shape[0]):
pval = [pool.apply(fisher_test, args = (row["module_genes"], 
gq_result_df.loc[j,"module_genes"], universeSize)) for j in range(i+1, gq_result_df.shape[0])]
#print("pval:", pval)
for j in range(i +1, fish_pval_mat.shape[0]):
pval_matrix[i, j] = pval[j -i -1]
pool.close()
pool.join()

运行外循环 119 次时的计算时间

1. 无并行化：13 分钟
2. 并行化(使用池(：12 分钟

如何优化并行化代码以减少时间？提前致谢

Answer 1

您的问题是使用Pool.apply()，因为它是一个阻塞调用。因此，您的执行不是并行的，而是顺序的。Pool.apply(( 块，直到结果可用，这使得这只是你提到的嵌套循环的另一个实现。您将一个块提交到子流程，等待它被处理，然后提交另一个 - 而不是一次性提交它们。

我不熟悉这种特定的算法，也不确定您是否可以并行化它 - 即块是要独立处理的，还是先前块的结果会影响连续任务，在这种情况下，这不会并行化。

如果它确实并行化，您可以尝试apply_async()。 如果这样做，则界面会略有变化，因为pval不再是结果列表，而是AsyncResult对象列表，您需要遍历这些对象并get()工作线程的实际结果。