使用Python多处理解决尴尬的并行问题 [英] Solving embarassingly parallel problems using Python multiprocessing
本文介绍了使用Python多处理解决尴尬的并行问题的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！
问题描述
尴尬的并行问题通常包括三个基本部分：
- 读取输入数据（来自文件，数据库，tcp连接等）。
- 对输入数据执行计算，其中每个计算独立于任何其他计算。
- 写入计算结果（到文件，数据库，tcp连接等）。
并行化程序在两个维度：
- 第2部分可以运行在多个内核，因为每个计算是独立的;
- 每个部件都可以独立运行。第1部分可以将数据放在输入队列中，第2部分可以从输入队列中取出数据并将结果放到输出队列中，第3部分可以将结果从输出队列中取出并写出。
这似乎是并发编程中最基本的模式，但我仍然在尝试解决它，因此让我们写一个规范的例子来说明这是如何使用多处理完成。
以下是示例问题：给定 CSV文件，以整数行作为输入，计算其总和。将问题分为三个部分，可以并行运行：
- 将输入文件处理为原始数据（整数的列表/
- 并行计算数据的总和
- 输出总和
下面是传统的单进程绑定Python程序，它解决了这三个任务：
＃！ usr / bin / env python
＃ - * - 编码：UTF-8 - * -
＃basicsums.py
一个程序从CSV文件读取整数值并写出他们的
总和到另一个CSV文件
import csv
import optparse
import sys
def make_cli_parser （）：
命令行接口解析器
usage =\\\
\\\
.join（[python％prog INPUT_CSV OUTPUT_CSV，
__doc__，
ARGUMENTS：
INPUT_CSV：一个输入CSV文件，其中包含数字行
OUTPUT_CSV：一个输出文件，其中包含sums \
]）
cli_parser = optparse.OptionParser（usage）
return cli_parser
def parse_input_csv（csvfile）：
CSV并产生元组，行的索引
作为第一个元素，行的整数作为第二个
元素。
索引是基于零索引的。
：参数：
- `csvfile`：a`csv.reader`实例
对于i，enumerate ）：
row = [int（entry）for row in row]
yield i，row
def sum_rows（rows）：
产生一个元组，每个输入整数列表的索引
作为第一个元素，整数列表的总和作为
第二个元素。
索引是基于零索引的。
：参数：
- `rows`：元组的可迭代，原始行的索引
作为第一个元素，整数列表作为第二个元素
for i，row in rows：
yield i，sum（row）
def write_results（csvfile，结果）：
将一系列结果写入outfile，其中第一列
是原始数据行的索引，第二列是
计算。
索引是基于零索引的。
：参数：
- `csvfile`：要写结果的`csv.writer`实例
- `results`：一个可迭代的元组，基于零的）
原始行作为第一个元素，计算结果
从该行作为第二个元素
for result_row结果：
csvfile.writerow（result_row）
def main（argv）：
cli_parser = make_cli_parser（）
opts，args = cli_parser.parse_args （argv）
如果len（args）！= 2：
cli_parser.error（请提供输入文件和输出文件。）
infile = open（args [0]）
in_csvfile = csv.reader（infile）
outfile = open（args [1]，'w'）
out_csvfile = csv.writer（outfile）
＃
input_rows = parse_input_csv（in_csvfile）
＃为可迭代的结果发送可迭代到sum_rows（）的行，但是
＃仍然未被求值
result_rows = sum_rows（input_rows）
＃final评估发生在一个链中write_results（）
write_results（out_csvfile，result_rows）
infile.close（）
outfile.close（）
if __name__ =='__main__'：
main（sys.argv [1：]）
让我们来接受这个程序，并重写它以使用多处理来并行化上述三个部分。下面是这个新的，并行化的程序的骨架，需要被充实以满足评论中的部分：
！/ usr / bin / env python
＃ - * - 编码：UTF-8 - * -
＃multroc_sums.py
一个程序从CSV文件读取整数值，写出他们的
总和到另一个CSV文件，如果需要使用多个进程。
import csv
import multiprocessing
import optparse
import sys
NUM_PROCS = multiprocessing.cpu_count（）
def make_cli_parser（）：
使命令行界面解析器
usage =\\\
\\\
.join（[python％prog INPUT_CSV OUTPUT_CSV，
__doc__，
自变量：
INPUT_CSV：一个包含数字行的输入CSV文件
OUTPUT_CSV：将包含sums \
的输出文件]）
cli_parser = optparse.OptionParser（用法）
cli_parser .add_option（' - n'，'--numprocs'，type ='int'，
default = NUM_PROCS，
help =要启动的进程数[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（请提供输入文件和输出文件）
infile = open（args [0]）
in_csvfile = csv.reader （infile）
outfile = open（args [1]，'w'）
out_csvfile = csv.writer（outfile）
＃解析输入文件并添加解析数据到
＃处理的队列，可能分块以减少
＃进程之间的通信。
＃使用与用户分配的进程一样多的进程，处理在
＃队列中任何（chunks）出现的解析数据
＃（opts.numprocs）;将结果放在队列中以输出。
＃
＃当解析器停止将数据放入
＃输入队列时终止进程。
＃当结果出现在输出
＃queue上时，将结果写入磁盘。
＃确保所有子进程都已终止。
＃清理文件。
infile.close（）
outfile.close（）
如果__name__ =='__main__'：
main（sys.argv [1： ]）
这些代码以及可生成示例CSV文件的另一段代码，以便进行测试，可以在github上找到。
我很感激这里有关并发大师会遇到这个问题。
这里有一些问题， >用于处理任何/全部的奖励积分：
- 我应该有子进程读取数据并将其放入队列，主进程这样做没有阻塞，直到所有输入都被读取为止？
- 同样，如果我有一个子进程从处理队列写出结果，或者主进程可以必须等待所有结果？
- 我应该使用进程池的和操作？
- 如果是，我在池上调用什么方法来让它开始处理进入输入队列的结果，而不阻塞输入和输出进程，太？ apply_async（）？ map_async（）？ imap（）？ imap_unordered（）？
- 假设我们不需要根据输入和输出队列抽取输入和输出队列，但可以等待所有输入都被解析，计算（例如，因为我们知道所有的输入和输出将适合系统存储器）。
解决方案如果我们以任何方式改变算法（例如， div>
我的解决方案有一个额外的响铃和哨声，以确保输出的顺序与输入的顺序相同。我使用multiprocessing.queue在进程之间发送数据，发送停止消息，以便每个进程知道退出检查队列。我认为源代码中的注释应该清楚发生了什么，但如果不让我知道。
＃！/ usr / bin / env python
＃ - * - 编码：UTF-8 - * -
＃multroc_sums.py
一个程序从CSV文件读取整数值， b $ b汇总到另一个CSV文件，如果需要使用多个进程。
import csv
import multiprocessing
import optparse
import sys
NUM_PROCS = multiprocessing.cpu_count（）
def make_cli_parser（）：
命令行界面解析器
用法=\\\
\\\
.join（[python％prog INPUT_CSV OUTPUT_CSV，
__doc__，
自变量：
INPUT_CSV：输入CSV文件
OUTPUT_CSV：一个包含sums \
的输出文件]
cli_parser = optparse.OptionParser（usage）
cli_parser.add_option（' -n'，'--numprocs'，type ='int'，
default = NUM_PROCS，
help =启动的进程数[DEFAULT：％default]）
return cli_parser
class CSVWorker（object）：
def __init __（self，numprocs，infile，outfile）：
self.numprocs = numprocs
self.infile = open ）
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（）
在self.ps中的p：
p.start（）
self.pin.join ）
i = 0
for self.ps：
p.join（）
printDone，i
i + = 1
self.pout.join（）
self.infile.close（）
def parse_input_csv（self）：
解析输入CSV并产生元组与行的索引
作为第一个元素，并将行的整数作为第二个
元素。
索引是基于零索引的。
然后通过队列发送数据以使工作者执行他们的
事情。最后，输入过程为每个
工作者发送一个STOP消息。
for i，row in enumerate（self.in_csvfile）：
row = [int（entry）for row in row]
self.inq.put（范围（self.numprocs）中的i的
：
self.inq.put（STOP）
def sum_row ：
工人。消耗inq并在outq上产生答案
tot = 0
for i，iter in iter（self.inq.get，STOP）：
self.outq .put（（i，sum（row）））
self.outq.put（STOP）
def write_output_csv（self）：
打开输出csv文件，然后开始读取outq的答案
由于我选择确保输出同步到输入
是一些额外的好东西。
显然，您的输入具有原始行号，因此这不是
必需的。
cur = 0
stop = 0
buffer = {}
＃出于某种原因，csv.writer在进程间工作不正常，因此打开/关闭
＃并在同一个进程中使用它，否则你会有最后
＃几行丢失
outfile = open（self.outfile，w）
self.out_csvfile = csv.writer（outfile）
#Keep运行，直到我们看到numprocs STOP消息
在范围内工作（self.numprocs）：
for i，val in iter .outq.get，STOP）：
＃如果不保存在缓冲区中，则验证行是否正确
如果i！= cur：
buffer [i] = val
else：
#if是写出来，并确保没有等待的行
self.out_csvfile.writerow（[i，val]）
cur + = 1
while cur在缓冲区中：
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）
如果len 2：
cli_parser.error（请提供输入文件和输出文件。）
c = CSVWorker（opts.numprocs，args [0]，args [1]）$ b
$ b if __name__ =='__main__'：
main（sys.argv [1：]）
How does one use multiprocessing to tackle embarrassingly parallel problems?
Embarassingly parallel problems typically consist of three basic parts:
- Read input data (from a file, database, tcp connection, etc.).
- Run calculations on the input data, where each calculation is independent of any other calculation.
- Write results of calculations (to a file, database, tcp connection, etc.).
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.
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.
Here is the example problem: Given a CSV file with rows of integers as input, compute their sums. Separate the problem into three parts, which can all run in parallel:
- Process the input file into raw data (lists/iterables of integers)
- Calculate the sums of the data, in parallel
- Output the sums
Below is traditional, single-process bound Python program which solves these three tasks:
#!/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 files for testing purposes, can be found on github.
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 pool for 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)?
解决方案
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.
#!/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:])
这篇关于使用Python多处理解决尴尬的并行问题的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
查看全文
