使用 Python 多处理解决令人尴尬的并行问题

Solving embarassingly parallel problems using Python multiprocessing

提问by gotgenes

How does one use multiprocessingto tackle embarrassingly parallel problems?

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

Embarassingly parallel problems typically consist of three basic parts:

令人尴尬的并行问题通常由三个基本部分组成：

1. Readinput data (from a file, database, tcp connection, etc.).
2. Runcalculations on the input data, where each calculation is independent of any other calculation.
3. Writeresults of calculations (to a file, database, tcp connection, etc.).

1. 读取输入数据（来自文件、数据库、tcp 连接等）。
2. 对输入数据运行计算，其中每个计算都独立于任何其他计算。
3. 将计算结果写入（到文件、数据库、tcp 连接等）。

We can parallelize the program in two dimensions:

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

• Part 2 can run on multiple cores, since each calculation is independent; order of processing doesn't matter.
• Each part can run independently. Part 1 can place data on an input queue, part 2 can pull data off the input queue and put results onto an output queue, and part 3 can pull results off the output queue and write them out.

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

This seems a most basic pattern in concurrent programming, but I am still lost in trying to solve it, so let's write a canonical example to illustrate how this is done using multiprocessing.

这似乎是并发编程中最基本的模式，但我仍然无法解决它，所以让我们编写一个规范的示例来说明如何使用 multiprocessing 完成此操作。

Here is the example problem: Given a CSV filewith rows of integers as input, compute their sums. Separate the problem into three parts, which can all run in parallel:

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

1. Process the input file into raw data (lists/iterables of integers)
2. Calculate the sums of the data, in parallel
3. Output the sums

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

Below is traditional, single-process bound Python program which solves these three tasks:

下面是传统的单进程绑定 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:])

Let's take this program and rewrite it to use multiprocessing to parallelize the three parts outlined above. Below is a skeleton of this new, parallelized program, that needs to be fleshed out to address the parts in the comments:

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

#!/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:])

These pieces of code, as well as another piece of code that can generate example CSV filesfor testing purposes, can be found on github.

可以在 github上找到这些代码段以及可以生成用于测试目的的示例 CSV 文件的另一段代码。

I would appreciate any insight here as to how you concurrency gurus would approach this problem.

如果您有任何关于并发专家将如何解决此问题的见解，我将不胜感激。

---

---

Here are some questions I had when thinking about this problem.Bonus points for addressing any/all:

以下是我在考虑这个问题时遇到的一些问题。解决任何/所有问题的奖励积分：

• Should I have child processes for reading in the data and placing it into the queue, or can the main process do this without blocking until all input is read?
• Likewise, should I have a child process for writing the results out from the processed queue, or can the main process do this without having to wait for all the results?
• Should I use a processes poolfor the sum operations?
  • If yes, what method do I call on the pool to get it to start processing the results coming into the input queue, without blocking the input and output processes, too? apply_async()? map_async()? imap()? imap_unordered()?
• Suppose we didn't need to siphon off the input and output queues as data entered them, but could wait until all input was parsed and all results were calculated (e.g., because we know all the input and output will fit in system memory). Should we change the algorithm in any way (e.g., not run any processes concurrently with I/O)?

• 我应该有子进程来读取数据并将其放入队列，还是主进程可以在不阻塞的情况下执行此操作，直到读取所有输入？
• 同样，我应该有一个子进程来从处理队列中写出结果，还是主进程可以这样做而不必等待所有结果？
• 我应该为求和操作使用进程池吗？
  • 如果是，我在池上调用什么方法来让它开始处理进入输入队列的结果，而不阻塞输入和输出进程？应用异步（）？地图异步（）？imap()? imap_unordered()?
• 假设我们不需要在数据进入时抽走输入和输出队列，而是可以等到所有输入被解析并计算所有结果（例如，因为我们知道所有输入和输出都适合系统内存）。我们是否应该以任何方式更改算法（例如，不要在 I/O 的同时运行任何进程）？

采纳答案by hbar

My solution has an extra bell and whistle to make sure that the order of the output has the same as the order of the input. I use multiprocessing.queue's to send data between processes, sending stop messages so each process knows to quit checking the queues. I think the comments in the source should make it clear what's going on but if not let me know.

我的解决方案有一个额外的花里胡哨，以确保输出的顺序与输入的顺序相同。我使用 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:])

回答by Gael Varoquaux

Coming late to the party...

聚会迟到了...

joblibhas a layer on top of multiprocessing to help making parallel for loops. It gives you facilities like a lazy dispatching of jobs, and better error reporting in addition to its very simple syntax.

joblib在多处理之上有一个层来帮助并行循环。除了非常简单的语法之外，它还为您提供了诸如延迟调度作业和更好的错误报告等功能。

As a disclaimer, I am the original author of joblib.

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

回答by Bogdan Kulynych

I realize that I'm a bit late for the party, but I've recently discovered GNU parallel, and want to show how easy it is to accomplish this typical task with it.

我意识到我参加聚会有点晚了，但我最近发现了GNU parallel，并且想展示用它完成这个典型任务是多么容易。

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

Something like this will do for sum.py:

像这样的事情会做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)

Parallel will run sum.pyfor every line in input.csv(in parallel, of course), then output the results to sums. Clearly better than multiprocessinghassle

Parallel 将对sum.py中的每一行运行input.csv（当然是并行的），然后将结果输出到sums. 显然比multiprocessing麻烦好

回答by S.Lott

Old School.

老套。

p1.py

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

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

p3.py

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

Here's the multi-processing final structure.

这是多处理最终结构。

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

Yes, the shell has knit these together at the OS level. It seems simpler to me and it works very nicely.

是的，外壳已经在操作系统级别将这些组合在一起。这对我来说似乎更简单，而且效果很好。

Yes, there's slightly more overhead in using pickle (or cPickle). The simplification, however, seems worth the effort.

是的，使用pickle（或cPickle）的开销略高。然而，这种简化似乎值得付出努力。

If you want the filename to be an argument to p1.py, that's an easy change.

如果您希望文件名成为 的参数p1.py，那很容易更改。

More importantly, a function like the following is very handy.

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

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

That allows you to do this:

这允许你这样做：

for item in get_stdin():
     process item

---

---

This is very simple, but it does not easilyallow you to have multiple copies of P2.py running.

这很简单，但它并不容易让您运行多个 P2.py 副本。

You have two problems: fan-out and fan-in. The P1.py must somehow fan out to multiple P2.py's. And the P2.py's must somehow merge their results into a single P3.py.

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

The old-school approach to fan-out is a "Push" architecture, which is very effective.

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

Theoretically, multiple P2.py's pulling from a common queue is the optimal allocation of resources. This is often ideal, but it's also a fair amount of programming. Is the programming really necessary? Or will round-robin processing be good enough?

理论上，多个P2.py从一个公共队列中拉取是资源的最优分配。这通常是理想的，但它也是大量的编程。编程真的有必要吗？或者循环处理是否足够好？

Practically, you'll find that making P1.py do a simple "round robin" dealing among multiple P2.py's may be quite good. You'd have P1.py configured to deal to ncopies of P2.py via named pipes. The P2.py's would each read from their appropriate pipe.

实际上，您会发现让 P1.py 在多个 P2.py 之间进行简单的“循环”处理可能非常好。您将 P1.py 配置为通过命名管道处理 P2.py 的n 个副本。P2.py's 将从各自的适当管道中读取。

What if one P2.py gets all the "worst case" data and runs way behind? Yes, round-robin isn't perfect. But it's better than only one P2.py and you can address this bias with simple randomization.

如果一个 P2.py 获得所有“最坏情况”数据并远远落后怎么办？是的，循环法并不完美。但它比只有一个 P2.py 好，您可以通过简单的随机化来解决这种偏差。

Fan-in from multiple P2.py's to one P3.py is a bit more complex, still. At this point, the old-school approach stops being advantageous. P3.py needs to read from multiple named pipes using the selectlibrary to interleave the reads.

从多个 P2.py 扇入到一个 P3.py 仍然有点复杂。在这一点上，老派方法不再具有优势。P3.py 需要使用select库从多个命名管道中读取以交错读取。

回答by Vatine

It's probably possible to introduce a bit of parallelism into part 1 as well. Probably not an issue with a format that's as simple as CSV, but if the processing of the input data is noticeably slower than the reading of the data, you could read larger chunks, then continue to read until you find a "row separator" (newline in the CSV case, but again that depends on the format read; doesn't work if the format is sufficiently complex).

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

These chunks, each probably containing multiple entries, can then be farmed off to a crowd of parallel processes reading jobs off a queue, where they're parsed and split, then placed on the in-queue for stage 2.

这些块，每个可能包含多个条目，然后可以被分配给一群并行进程，从队列中读取作业，在那里它们被解析和拆分，然后放置在第 2 阶段的队列中。