python多进程map比apply快_Python多处理队列比pool.map慢

我最近开始尝试使用多处理来加速任务.我创建了一个执行模糊字符串匹配的脚本,并使用不同的算法计算得分(我想比较不同的匹配技术).你可以在这里找到完整的源代码：

https://bitbucket.org/bergonzzi/fuzzy-compare/src.作为输入,它需要2个文件组合成对(file1的每一行与file2的每一行).对于每对,计算模糊匹配分数.

我做了3个版本.运行我的仓库中提供的样本数据(成对组合后包含697.340项),我有以下时间：

>简单的单一过程 – 0:00:47

>使用Pool.map()的多进程 – 0:00:13

>使用队列的多进程(生产者/消费者模式) – 0:01:04

我试图理解为什么我的Pool.map()版本比我的Queue版本快得多,这实际上比简单的单进程版本慢.

我甚至尝试使用Queues的原因是Pool.map()版本保留了结果,直到所有内容都完成并且只在最后写入文件.这意味着对于大文件,它最终会占用大量内存. I’m talking about this version(链接到它,因为这里要粘贴很多代码).

To solve this I refactored it into a producer/consumer pattern(或至少尝试过).在这里,我首先通过组合两个输入文件来生成作业,并将它们放入消费者处理的队列中(计算模糊匹配分数).完成的工作被放入队列中.然后我有一个进程从这个队列中抓取完成的项目并将它们写入文件.这样,从理论上讲,我不需要那么多的内存,因为结果会被刷新到磁盘上.它似乎工作正常,但它慢得多.我还注意到,当我在Mac OSX上查看Activity Monitor时,我正在产生的4个进程似乎没有消耗掉100％的CPU(这与Pool.map()版本不同).

我注意到的另一件事是我的生产者函数似乎正确地填满了队列,但是消费者进程似乎要等到队列被填满而不是在第一个项目到达时立即开始工作.我可能在那里做错了……

这里有一些关于Queue版本的相关代码的参考(尽管最好查看上面链接的repo中的完整代码).

这是我的生产者功能：

def combine(list1, list2):

'''

Combine every item of list1 with every item of list 2,

normalize put the pair in the job queue.

'''

pname = multiprocessing.current_process().name

for x in list1:

for y in list2:

# slugify is a function to normalize the strings

term1 = slugify(x.strip(), separator=' ')

term2 = slugify(y.strip(), separator=' ')

job_queue.put_nowait([term1, term2])

这是作家功能：

def writer(writer_queue):

out = open(file_out, 'wb')

pname = multiprocessing.current_process().name

out.write(header)

for match in iter(writer_queue.get, "STOP"):

print("%s is writing %s") % (pname, str(match))

line = str(';'.join(match) + 'n')

out.write(line)

out.close()

这是执行实际计算的worker函数(剥离了大部分代码,因为它在这里没有区别,repo上的完整源代码)：

def score_it(job_queue, writer_queue):

'''Calculate scores for pair of words.'''

pname = multiprocessing.current_process().name

for pair in iter(job_queue.get_nowait, "STOP"):

# do all the calculations and put the result into the writer queue

writer_queue.put(result)

这是我设置流程的方式：

# Files

to_match = open(args.file_to_match).readlines()

source_list = open(args.file_to_be_matched).readlines()

workers = 4

job_queue = multiprocessing.Manager().Queue()

writer_queue = multiprocessing.Manager().Queue()

processes = []

print('Start matching with "%s", minimum score of %s and %s workers') % (

args.algorithm, minscore, workers)

# Fill up job queue

print("Filling up job queue with term pairs...")

c = multiprocessing.Process(target=combine, name="Feeder", args=(to_match, source_list))

c.start()

c.join()

print("Job queue size: %s") % job_queue.qsize()

# Start writer process

w = multiprocessing.Process(target=writer, name="Writer", args=(writer_queue,))

w.start()

for w in xrange(workers):

p = multiprocessing.Process(target=score_it, args=(job_queue, writer_queue))

p.start()

processes.append(p)

job_queue.put("STOP")

for p in processes:

p.join()

writer_queue.put("STOP")

我在这里读到了很多关于多处理有时候比较慢的事情,我知道这与创建和管理新进程的开销有关.此外,当要完成的工作不够“大”时,多处理的效果可能不可见.但是在这种情况下,我认为这个工作非常大,而且Pool.map()版本似乎也证明了这一点,因为它的速度要快得多.

在管理所有这些进程并传递队列对象时,我是否做了一些非常错误的事情？如何对其进行优化,以便在处理文件时将结果写入文件,以便最大限度地减少运行时所需的内存量？

谢谢！