函数内部的并行性？

Question

我有一个函数，它计算项目列表在下面rows中出现的频率：

def count(pair_list):
    return float(sum([1 for row in rows if all(item in row.split() for item in pair_list)]))

if __name__ == "__main__":
    pairs = [['apple', 'banana'], ['cookie', 'popsicle'], ['candy', 'cookie'], ...]
    # grocery transaction data
    rows = ['apple cookie banana popsicle wafer', 'almond milk eggs butter bread', 'bread almonds apple', 'cookie candy popsicle pop', ...]

    res = [count(pair) for pair in pairs]

实际上，len(rows)是10000，并且pairs中有18000个元素，因此count()中的列表理解和主函数中的列表理解的计算成本是昂贵的。你知道吗

我尝试了一些并行处理：

from multiprocessing.dummy import Pool as ThreadPool
import multiprocessing as mp

threadpool = ThreadPool(processes = mp.cpu_count())

res = threadpool.map(count, pairs)

这也不是很快。事实上，15分钟后，我就辞职了，因为看起来还没有结束。两个问题：1）如何加快在count()中进行的实际搜索？2） 如何检查threadpool.map进程的状态（即，查看还有多少对需要迭代）？你知道吗

Answer 1

1）计算的总体复杂性是巨大的，它来自不同的来源：

a）在较低的计算级别上拆分行，因此python必须为每个迭代创建新的行拆分。为了避免这种情况，可以预先计算行。类似这样的操作可以完成这项工作（对“count”函数做一些小的更改）：

rows2 = [row.split() for row in rows]

b）你逐个比较列表项，即使你只需要检查另一个列表中是否有单词。在这里我们可以对它进行更多的调整（在“count”函数中使用rows3而不是rows2）：

rows3 = [set(row.split()) for row in rows]

def count(pair_list):
    return float(sum([1 for row in rows3 if all(item in row for item in pair_list)]))

c）你把每一个单词成对检查，每一个单词排成一行。对于原始版本，每次调用“count”函数，计算需要2*len（row）*len（rows）次迭代，但所需时间可能更少。对于选项b），在好的情况下可以减少到2*len（行），但是可以对每对进行一组查找，而不是2组。 诀窍是为每一行准备所有可能的单词*单词组合，并检查这个集合中是否存在相应的元组。 因此，在main函数中创建复杂的不可变搜索结构：

rows4 = [set((a, b) for a in row for b in row) for row in rows2]

现在“count”看起来不同了，它需要tuple而不是list：

def count2(pair):
    return float(len([1 for row in rows4 if(pair in row)]))

所以你叫它有点不同： res=[count2（tuple（pair））表示成对]

请注意，搜索结构的创建需要len(行。拆分（））^2每行的时间和空间，所以如果您的行可以长，它不是最佳的。毕竟，选项b）可以更好。你知道吗

2）您可以预测“count”的呼叫数-它是len（pairs）。对“Count”函数的调用进行计数，并在其中进行调试打印，例如，每1000次调用。你知道吗