Решение неприятно параллельных проблем с использованием многопроцессорной обработки 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 .

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

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

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

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

Опоздал на вечеринку ...

В joblib есть слой поверх многопроцессорной обработки, который помогает выполнять параллельные циклы for. В дополнение к очень простому синтаксису он предоставляет такие возможности, как отложенная отправка заданий и лучший отчет об ошибках.

В качестве оговорки, я являюсь первоначальным автором joblib.

Я понимаю, что немного опаздываю на вечеринку, но я недавно обнаружил GNU parallel и хочу показать, насколько легко с ним выполнить эту типичную задачу.

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суеты

Старая школа.

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.

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

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

На практике вы обнаружите, что заставить P1.py выполнять простую «циклическую» работу с несколькими P2.py может быть неплохо. У вас должен быть P1.py, настроенный для работы с n копиями P2.py через именованные каналы. Каждый P2.py будет читать из своего канала.

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

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

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

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