python - 使用Python多处理解决令人难以置信的并行问题

如何使用multiprocessing来处理embarrassingly parallel problems
令人尴尬的平行问题通常由三个基本部分组成:
读取输入数据(从文件、数据库、TCP连接等)。
对输入数据运行计算,其中每个计算独立于任何其他计算。
将计算结果写入(文件、数据库、TCP连接等)。
我们可以在两个维度上并行程序:
第2部分可以在多个内核上运行,因为每个计算都是独立的;处理顺序无关紧要。
每个部分都可以独立运行。第1部分可以将数据放在输入队列上,第2部分可以将数据从输入队列中拉出来并将结果放在输出队列上,第3部分可以将结果从输出队列中拉出来并将其写出。
这似乎是并发编程中最基本的模式,但我仍然无法解决它,所以让我们编写一个规范的示例来说明如何使用多处理来完成这一点。
下面是示例问题:给定一个以整数行作为输入的CSV file计算它们的和。将问题分为三个部分,这些部分可以并行运行:
将输入文件处理为原始数据(整数的列表/i数组)
并行计算数据的总和
输出总和
下面是传统的、单进程绑定的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:])

这些代码以及用于测试目的的another piece of code that can generate example CSV files可以是found on github
对于您如何处理这个问题,我将非常感谢您的任何见解。
以下是我在考虑这个问题时遇到的一些问题。解决任何/所有问题的奖励积分:
我应该有子进程来读取数据并将其放入队列中,还是主进程可以在读取所有输入之前不阻塞地执行此操作?
同样,我应该有一个子进程来从已处理的队列中写出结果,还是主进程可以这样做而不必等待所有结果?
我应该使用aprocesses pool进行求和操作吗?
如果是,我应该调用什么方法来让池在不阻塞输入和输出进程的情况下开始处理进入输入队列的结果?apply_async()map_async()imap()imap_unordered()
假设我们不需要在数据输入时从输入和输出队列中吸取数据,但是可以等到所有输入都被解析并计算出所有结果(例如,因为我们知道所有输入和输出都将适合系统内存)。我们是否应该以任何方式更改算法(例如,不与I/O同时运行任何进程)?

最佳答案

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

本文翻译自 https://stackoverflow.com/questions/2359253/

网站遵循 CC BY-SA 4.0 协议,转载或引用请注明出处。

标签 python concurrency multiprocessing embarrassingly-parallel


相关文章:

c++ - 为此有互锁吗? C ++

c++ - 如何从另一个线程中引发的Boost信号更新Qt GUI?

python - 多处理模块和pyro的比较?

python - Scikit-learn是否发布python GIL?

python - 在IIS7 64位上安装Mercurial时出错:“ DLL加载失败:%1不是有效的Win32应用程序”

java - 在并发交换期间隔离有状态Bean的最正确方法

python - 在同一对象上对多个函数进行单独处理的最有效方法

c - 使用大规模CPU的多线程与多处理

python - 为什么我们需要检查value是否等于None?

python - 计算具有1000位数字的第一个斐波那契数(项目Euler#25)