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

如何使用 多重处理来处理 令人尴尬的平行问题？

令人尴尬的平行问题通常包括三个基本部分:

  1. 读取 输入数据(从文件、数据库、 tcp 连接等)。
  2. 对输入数据运行 计算，其中每个计算为 独立于任何其他计算。
  3. 写入 计算结果(到文件、数据库、 tcp 连接等)。

我们可以将程序并行化为两个维度:

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

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

下面是示例问题: 给定一个以整数行作为输入的 CSV 档案，计算它们的总和。把问题分成三个部分，这三个部分可以并行运行:

  1. 将输入文件处理为原始数据(列出/迭代整数)
  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:])

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

如果您对并发专家如何处理这个问题有任何见解，我将不胜感激。

以下是我在思考这个问题时提出的一些问题:

  • 我是否应该有子进程来读取数据并将其放入队列，或者主进程可以在读取所有输入之前不受阻塞地完成这项工作？
  • 同样，我是否应该有一个子进程来从处理过的队列中写出结果，或者主进程可以不必等待所有结果就完成这项工作吗？
  • 我是否应该使用 工序池进行求和运算？
  • 假设我们不需要在数据输入时抽取输入和输出队列，而是可以等到解析了所有输入并计算了所有结果(例如，因为我们知道所有的输入和输出将适合系统内存)。我们是否应该以任何方式更改算法(例如，不与 I/O 并发运行任何进程) ？
20629 次浏览
小开

Old School.

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

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.

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

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 easily allow you to have multiple copies of P2.py running.

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.

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?

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 n copies of P2.py via named pipes. The P2.py's would each read from their appropriate pipe.

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.

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 select library to interleave the reads.

小开
最佳答案

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:])
小开

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).

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.

小开

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.

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

Something like this will do for 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.py for every line in input.csv (in parallel, of course), then output the results to sums. Clearly better than multiprocessing hassle

小开

Coming late to the party...

joblib has 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.

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


提交答案