Решение смущающе параллельных проблем с помощью многопроцессорной обработки Python

Question

Как можно использовать многопроцессорность для решения смущающих параллельных проблем ?

Параллельно смущающие проблемы обычно состоят из трех основных частей:

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 .

Буду признателен за понимание того, как вы, гуру параллелизма, подходите к этой проблеме.

<ч />

Вот несколько вопросов, которые у меня возникли, когда я думал об этой проблеме. Бонусные баллы за решение любых / всех вопросов:

• Должны ли у меня быть дочерние процессы для чтения данных и помещения их в очередь, или основной процесс может делать это без блокировки, пока не будет прочитан весь ввод?
• Аналогично, должен ли я иметь дочерний процесс для записи результатов из обработанной очереди, или основной процесс может сделать это без необходимости ждать всех результатов?
• Должен ли я использовать пул процессов для операций суммы?
  • Если да, какой метод я вызову в пуле, чтобы он начал обрабатывать результаты, поступающие во входную очередь, также не блокируя процессы ввода и вывода? apply_async () ? map_async () ? IMAP () ? imap_unordered ()
• Предположим, нам не нужно откачивать входные и выходные очереди при вводе данных, но мы можем подождать, пока все входные данные будут проанализированы и все результаты вычислены (например, потому что мы знаем, что все входные и выходные данные поместятся в систему объем памяти). Должны ли мы каким-либо образом изменить алгоритм (например, не запускать какие-либо процессы одновременно с вводом / выводом)?

Answer 1

Мое решение имеет дополнительный звонок и свисток, чтобы убедиться, что порядок вывода совпадает с порядком ввода. Я использую 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)

Parallel будет запускать sum.py для каждой строки в input.csv (конечно, параллельно), а затем выводить результаты в 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

Да, оболочка связала их вместе на уровне ОС. Это кажется мне проще и работает очень хорошо.

Да, использование pickle (или cPickle) немного больше. Однако упрощение, похоже, стоит усилий.

Если вы хотите, чтобы имя файла было аргументом p1.py, это легко изменить.

Что еще более важно, функция, подобная следующей, очень удобна.

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.

Подход старой школы к разветвлению - это архитектура "Push", которая очень эффективна.

Теоретически, множественное извлечение P2.py из общей очереди является оптимальным распределением ресурсов. Это часто идеально, но это также изрядное количество программирования. Программирование действительно необходимо? Или будет достаточно хорошая круговая обработка?

Практически, вы обнаружите, что заставить P1.py выполнять простой "круговой прием", работающий с несколькими P2.py, может быть неплохо. Вы должны настроить P1.py для работы с n копиями P2.py через именованные каналы. Каждый из P2.py читал бы из соответствующей трубы.

Что если один P2.py получит все данные в «наихудшем случае» и отстанет? Да, круговой прием не идеален. Но это лучше, чем один P2.py, и вы можете решить эту проблему с помощью простой рандомизации.

Разветвление от нескольких P2.py к одному P3.py все же немного сложнее. На этом этапе подход старой школы перестает быть выгодным. P3.py должен читать из нескольких именованных каналов, используя библиотеку select для чередования чтений.

Answer 5

Вероятно, можно ввести немного параллелизма и в первую часть. Вероятно, это не проблема с форматом, который так прост, как CSV, но если обработка входных данных заметно медленнее, чем чтение данных, вы можете прочитать более крупные фрагменты, а затем продолжить чтение, пока не найдете «разделитель строк» ​​( символ новой строки в случае CSV, но, опять же, это зависит от прочитанного формата; не работает, если формат достаточно сложный).

Эти чанки, каждый из которых, вероятно, содержит несколько записей, затем могут быть переданы в толпу параллельных процессов, читающих задания из очереди, где они анализируются и разделяются, а затем помещаются в очередь на этапе 2.