使用Python多处理解决不平行的问题

如何使用多处理来解决尴尬的并行问题 ？

不解之缘的并行问题通常由三个基本部分组成：

1. 读取input数据（从文件，数据库，TCP连接等）。
2. 对input数据运行计算，每个计算独立于任何其他计算 。
3. 写入计算结果（到文件，数据库，tcp连接等）。

我们可以在两个维度上并行化程序：

• 第2部分可以在多个核上运行，因为每个计算都是独立的; 处理顺序无关紧要。
• 每个部分都可以独立运行。 第1部分可以将数据放在input队列中，第2部分可以将数据从input队列中取出，并将结果放到输出队列中，第3部分可以将结果从输出队列中取出并写出。

这似乎是并发编程的一个最基本的模式，但是我仍然试图解决它，所以让我们编写一个规范的例子来说明如何使用多处理来完成 。

以下是示例问题：给定一个包含整数行作为input的CSV文件 ，计算它们的总和。 将问题分成三部分，可以并行运行：

1. 将input文件处理成原始数据（列表/整数的迭代）
2. 并行计算数据的总和
3. 输出总和

下面是传统的，单进程绑定的Python程序，它解决了这三个任务：

#!/usr/bin/env python # -*- coding: UTF-8 -*- # basicsums.py """A program that reads integer values from a CSV file and writes out their sums to another CSV file. """ import csv import optparse import sys def make_cli_parser(): """Make the command line interface parser.""" usage = "\n\n".join(["python %prog INPUT_CSV OUTPUT_CSV", __doc__, """ ARGUMENTS: INPUT_CSV: an input CSV file with rows of numbers OUTPUT_CSV: an output file that will contain the sums\ """]) cli_parser = optparse.OptionParser(usage) return cli_parser def parse_input_csv(csvfile): """Parses the input CSV and yields tuples with the index of the row as the first element, and the integers of the row as the second element. The index is zero-index based. :Parameters: - `csvfile`: a `csv.reader` instance """ for i, row in enumerate(csvfile): row = [int(entry) for entry in row] yield i, row def sum_rows(rows): """Yields a tuple with the index of each input list of integers as the first element, and the sum of the list of integers as the second element. The index is zero-index based. :Parameters: - `rows`: an iterable of tuples, with the index of the original row as the first element, and a list of integers as the second element """ for i, row in rows: yield i, sum(row) def write_results(csvfile, results): """Writes a series of results to an outfile, where the first column is the index of the original row of data, and the second column is the result of the calculation. The index is zero-index based. :Parameters: - `csvfile`: a `csv.writer` instance to which to write results - `results`: an iterable of tuples, with the index (zero-based) of the original row as the first element, and the calculated result from that row as the second element """ for result_row in results: csvfile.writerow(result_row) def main(argv): cli_parser = make_cli_parser() opts, args = cli_parser.parse_args(argv) if len(args) != 2: cli_parser.error("Please provide an input file and output file.") infile = open(args[0]) in_csvfile = csv.reader(infile) outfile = open(args[1], 'w') out_csvfile = csv.writer(outfile) # gets an iterable of rows that's not yet evaluated input_rows = parse_input_csv(in_csvfile) # sends the rows iterable to sum_rows() for results iterable, but # still not evaluated result_rows = sum_rows(input_rows) # finally evaluation takes place as a chain in write_results() write_results(out_csvfile, result_rows) infile.close() outfile.close() if __name__ == '__main__': main(sys.argv[1:]) 

让我们来看看这个程序，并重写它以使用多处理来并行化上面概述的三个部分。 下面是这个新的并行化程序的框架，需要充实以解决评论中的部分内容：

 #!/usr/bin/env python # -*- coding: UTF-8 -*- # multiproc_sums.py """A program that reads integer values from a CSV file and writes out their sums to another CSV file, using multiple processes if desired. """ import csv import multiprocessing import optparse import sys NUM_PROCS = multiprocessing.cpu_count() def make_cli_parser(): """Make the command line interface parser.""" usage = "\n\n".join(["python %prog INPUT_CSV OUTPUT_CSV", __doc__, """ ARGUMENTS: INPUT_CSV: an input CSV file with rows of numbers OUTPUT_CSV: an output file that will contain the sums\ """]) cli_parser = optparse.OptionParser(usage) cli_parser.add_option('-n', '--numprocs', type='int', default=NUM_PROCS, help="Number of processes to launch [DEFAULT: %default]") return cli_parser def main(argv): cli_parser = make_cli_parser() opts, args = cli_parser.parse_args(argv) if len(args) != 2: cli_parser.error("Please provide an input file and output file.") infile = open(args[0]) in_csvfile = csv.reader(infile) outfile = open(args[1], 'w') out_csvfile = csv.writer(outfile) # Parse the input file and add the parsed data to a queue for # processing, possibly chunking to decrease communication between # processes. # Process the parsed data as soon as any (chunks) appear on the # queue, using as many processes as allotted by the user # (opts.numprocs); place results on a queue for output. # # Terminate processes when the parser stops putting data in the # input queue. # Write the results to disk as soon as they appear on the output # queue. # Ensure all child processes have terminated. # Clean up files. infile.close() outfile.close() if __name__ == '__main__': main(sys.argv[1:]) 

这些代码片段，以及另一段可以生成用于testing目的的CSV文件的代码，可以 在github上find 。

如果您对并发性大师会如何解决这个问题，我将非常感激。

---

以下是我在思考这个问题时遇到的一些问题。 附加点解决任何/所有：

• 我是否应该让subprocess读取数据并将其放入队列，或者主进程是否可以在不读取所有input的情况下执行此操作？
• 同样，我是否应该有一个从处理队列中写出结果的subprocess，或者主进程是否可以执行此操作，而无需等待所有结果？
• 我应该使用进程池进行总和操作吗？
  • 如果是的话，我会调用哪个方法让它开始处理进入input队列的结果，而不会阻塞input和输出进程呢？ apply_async（） ？ map_async（） ？ imap（） ？ imap_unordered（） ？
• 假设我们不需要在input和输出队列中inputinput和输出队列，但是可以等到所有input被parsing并计算出所有结果（例如，因为我们知道所有input和输出都适合系统内存）。 我们应该以任何方式改变algorithm（例如，不要与I / O同时运行任何进程）？

Answer 1

我的解决scheme有一个额外的钟声和哨声，以确保输出的顺序与input的顺序相同。 我使用multiprocessing.queue在进程之间发送数据，发送停止消息，以便每个进程知道要退出检查队列。 我认为来源方面的意见应该清楚发生了什么，但是如果不让我知道的话。

 #!/usr/bin/env python # -*- coding: UTF-8 -*- # multiproc_sums.py """A program that reads integer values from a CSV file and writes out their sums to another CSV file, using multiple processes if desired. """ import csv import multiprocessing import optparse import sys NUM_PROCS = multiprocessing.cpu_count() def make_cli_parser(): """Make the command line interface parser.""" usage = "\n\n".join(["python %prog INPUT_CSV OUTPUT_CSV", __doc__, """ ARGUMENTS: INPUT_CSV: an input CSV file with rows of numbers OUTPUT_CSV: an output file that will contain the sums\ """]) cli_parser = optparse.OptionParser(usage) cli_parser.add_option('-n', '--numprocs', type='int', default=NUM_PROCS, help="Number of processes to launch [DEFAULT: %default]") return cli_parser class CSVWorker(object): def __init__(self, numprocs, infile, outfile): self.numprocs = numprocs self.infile = open(infile) self.outfile = outfile self.in_csvfile = csv.reader(self.infile) self.inq = multiprocessing.Queue() self.outq = multiprocessing.Queue() self.pin = multiprocessing.Process(target=self.parse_input_csv, args=()) self.pout = multiprocessing.Process(target=self.write_output_csv, args=()) self.ps = [ multiprocessing.Process(target=self.sum_row, args=()) for i in range(self.numprocs)] self.pin.start() self.pout.start() for p in self.ps: p.start() self.pin.join() i = 0 for p in self.ps: p.join() print "Done", i i += 1 self.pout.join() self.infile.close() def parse_input_csv(self): """Parses the input CSV and yields tuples with the index of the row as the first element, and the integers of the row as the second element. The index is zero-index based. The data is then sent over inqueue for the workers to do their thing. At the end the input process sends a 'STOP' message for each worker. """ for i, row in enumerate(self.in_csvfile): row = [ int(entry) for entry in row ] self.inq.put( (i, row) ) for i in range(self.numprocs): self.inq.put("STOP") def sum_row(self): """ Workers. Consume inq and produce answers on outq """ tot = 0 for i, row in iter(self.inq.get, "STOP"): self.outq.put( (i, sum(row)) ) self.outq.put("STOP") def write_output_csv(self): """ Open outgoing csv file then start reading outq for answers Since I chose to make sure output was synchronized to the input there is some extra goodies to do that. Obviously your input has the original row number so this is not required. """ cur = 0 stop = 0 buffer = {} # For some reason csv.writer works badly across processes so open/close # and use it all in the same process or else you'll have the last # several rows missing outfile = open(self.outfile, "w") self.out_csvfile = csv.writer(outfile) #Keep running until we see numprocs STOP messages for works in range(self.numprocs): for i, val in iter(self.outq.get, "STOP"): # verify rows are in order, if not save in buffer if i != cur: buffer[i] = val else: #if yes are write it out and make sure no waiting rows exist self.out_csvfile.writerow( [i, val] ) cur += 1 while cur in buffer: self.out_csvfile.writerow([ cur, buffer[cur] ]) del buffer[cur] cur += 1 outfile.close() def main(argv): cli_parser = make_cli_parser() opts, args = cli_parser.parse_args(argv) if len(args) != 2: cli_parser.error("Please provide an input file and output file.") c = CSVWorker(opts.numprocs, args[0], args[1]) if __name__ == '__main__': main(sys.argv[1:]) 

Answer 2

来晚了…

joblib在多处理之上有一个层，可以帮助并行处理循环。 它为您提供了懒惰的调度工作，除了非常简单的语法外，还提供了更好的错误报告。

作为一个免责声明，我是joblib的原始作者。

Answer 3

我意识到我晚了一点，但最近我发现了GNU并行 ，并且想要展示如何用它来完成这个典型的任务。

 cat input.csv | parallel ./sum.py --pipe > sums 

像这样的东西将sum.py ：

 #!/usr/bin/python from sys import argv if __name__ == '__main__': row = argv[-1] values = (int(value) for value in row.split(',')) print row, ':', sum(values) 

并行将在sum.py中的每一行运行sum.py （当然是并行），然后输出结果到sums 。 明显好于multiprocessing麻烦

Answer 4

老套。

p1.py

 import csv import pickle import sys with open( "someFile", "rb" ) as source: rdr = csv.reader( source ) for line in eumerate( rdr ): pickle.dump( line, sys.stdout ) 

p2.py

 import pickle import sys while True: try: i, row = pickle.load( sys.stdin ) except EOFError: break pickle.dump( i, sum(row) ) 

p3.py

 import pickle import sys while True: try: i, row = pickle.load( sys.stdin ) except EOFError: break print i, row 

这是多处理的最终结构。

 python p1.py | python p2.py | python p3.py 

是的，shell已经在操作系统级别一起编织。 这对我来说似乎更简单，而且工作得非常好。

是的，在使用pickle（或者cPickle）的时候会有更多的开销。 但是，简化似乎值得付出努力。

如果你想把文件名作为p1.py一个参数，这是一个简单的改变。

更重要的是，像下面这样的function非常方便。

 def get_stdin(): while True: try: yield pickle.load( sys.stdin ) except EOFError: return 

这可以让你做到这一点：

 for item in get_stdin(): process item 

---

这很简单，但是不容易让你有多个P2.py副本运行。

你有两个问题：扇出和扇入。 P1.py必须以某种方式扇出多个P2.py。 而P2.py必须以某种方式将它们的结果合并到一个P3.py中。

老派的扇出方法是“推”式架构，非常有效。

从理论上讲，多个P2.py从普通队列中拉出是资源的最优分配。 这通常是理想的，但也是相当多的编程。 编程是否真的有必要？ 或者循环加工是否足够好？

实际上，你会发现，使P1.py做一个简单的“循环”处理多个P2.py可能是相当不错的。 您可以将P1.pyconfiguration为通过命名pipe道处理n个P2.py副本。 P2.py将分别从相应的pipe道读取。

如果一个P2.py得到所有“最坏情况”的数据并且落后于什么呢？ 是的，循环法并不完美。 但是它比只有一个P2.py更好，你可以用简单的随机来解决这个偏见。

从多个P2.py的扇入到一个P3.py有点复杂。 在这一点上，老派的做法不再有利。 P3.py需要使用select库从多个命名pipe道读取交错读取。

Answer 5

也许可以在第一部分中引入一些并行性。 可能不是像CSV那样简单的格式问题，但是如果input数据的处理明显慢于读取数据，则可以读取更大的块，然后继续读取，直到find“行分隔符”（ CSV格式的换行符，但是这又取决于读取的格式;如果格式足够复杂，则不起作用）。

这些块可能包含多个条目，然后可以分散到大量的并行进程中，将作业从队列中读取出来，在队列中进行分析和拆分，然后放到第2阶段的队列中。