Решение проблем с параллельными задачами с использованием многопроцессорной обработки Python

Question

Как использовать multiprocessing для решения проблемы embarrassingly parallel problems?

чрезвычайная параллельности проблема, как правило, состоит из трех основных частей:

1. Чтение входных данных (из файла, баз данных, соединение TCP и т.д.).
2. Выполнение расчетов по входным данным, где каждый расчет равен независимо от каких-либо других расчетов.
3. Напишите результаты расчетов (в файл, базу данных, подключение к tcp и т. Д.).

Мы можем распараллелить программу в двух измерениях:

• Часть 2 может работать на нескольких ядрах, поскольку каждый расчет не зависит; порядок обработки не имеет значения.
• Каждая часть может работать независимо. Часть 1 может помещать данные в очередь ввода, часть 2 может вытащить данные из входной очереди и поместить результаты в очередь вывода, а часть 3 может вывести результаты из очереди вывода и записать их.

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

Вот пример проблемы: Учитывая CSV file с целыми числами в качестве входных данных, вычислите их суммы. Отдельные проблемы на три части, которые все могут работать параллельно:

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

Эти фрагменты кода, а также another piece of code that can generate example CSV files для целей тестирования, может быть found on github.

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

---

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

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

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

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

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 

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

Да, есть немного больше накладных расходов при использовании соленья (или 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's должны каким-то образом объединить свои результаты в одну P3.py.

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

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

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

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

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

Answer 3

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

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

Answer 4

Я понимаю, что я немного опаздываю на вечеринку, но недавно я обнаружил 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) 

Параллельно будет работать sum.py для каждой строки в input.csv (параллельно, конечно), то выходные результаты в sums. Ясно, что лучше, чем multiprocessing хлопот

Answer 5

опоздание к партии ...

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

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