Resolver problemas embarazosamente paralelos usando el multiprocesamiento de Python

¿Cómo se utiliza el multiprocesamiento para abordar problemas embarazosamente paralelos ?

Los problemas vergonzosamente paralelos suelen constar de tres partes básicas:

1. Leer datos de entrada (de un archivo, base de datos, conexión tcp, etc.).
2. Ejecute cálculos sobre los datos de entrada, donde cada cálculo es independiente de cualquier otro cálculo .
3. Escribe los resultados de los cálculos (en un archivo, base de datos, conexión tcp, etc.).

Podemos paralelizar el programa en dos dimensiones:

• La parte 2 se puede ejecutar en varios núcleos, ya que cada cálculo es independiente; el orden de procesamiento no importa.
• Cada parte puede funcionar de forma independiente. La parte 1 puede colocar datos en una cola de entrada, la parte 2 puede sacar datos de la cola de entrada y poner los resultados en una cola de salida, y la parte 3 puede sacar resultados de la cola de salida y escribirlos.

Este parece un patrón más básico en la programación concurrente, pero todavía estoy perdido tratando de resolverlo, así que escribamos un ejemplo canónico para ilustrar cómo se hace esto usando multiprocesamiento .

Este es el problema de ejemplo: dado un archivo CSV con filas de números enteros como entrada, calcule sus sumas. Separe el problema en tres partes, que pueden ejecutarse en paralelo:

1. Procesar el archivo de entrada en datos brutos (listas / iterables de enteros)
2. Calcule las sumas de los datos, en paralelo
3. Salida de las sumas

A continuación se muestra un programa tradicional de Python vinculado a un solo proceso que resuelve estas tres tareas:

#!/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:])

Tomemos este programa y reescribamoslo para usar multiprocesamiento para paralelizar las tres partes descritas anteriormente. A continuación se muestra un esqueleto de este nuevo programa paralelo, que debe desarrollarse para abordar las partes en los comentarios:

#!/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:])

Estos fragmentos de código, así como otro fragmento de código que puede generar archivos CSV de ejemplo con fines de prueba, se pueden encontrar en github .

Agradecería cualquier idea aquí sobre cómo los gurús de la concurrencia abordarían este problema.

Aquí hay algunas preguntas que tuve al pensar en este problema. Puntos de bonificación por abordar cualquiera / todos:

• ¿Debería tener procesos secundarios para leer los datos y colocarlos en la cola, o puede el proceso principal hacer esto sin bloquear hasta que se lean todas las entradas?
• Del mismo modo, ¿debería tener un proceso secundario para escribir los resultados de la cola procesada, o el proceso principal puede hacer esto sin tener que esperar todos los resultados?
• ¿Debo usar un grupo de procesos para las operaciones de suma?
  • Si es así, ¿a qué método llamo en el grupo para que comience a procesar los resultados que ingresan a la cola de entrada, sin bloquear también los procesos de entrada y salida? apply_async () ? map_async () ? imap () ? imap_unordered () ?
• Supongamos que no necesitamos desviar las colas de entrada y salida a medida que los datos ingresan, pero podríamos esperar hasta que se analizaran todas las entradas y se calcularan todos los resultados (por ejemplo, porque sabemos que todas las entradas y salidas caben en la memoria del sistema). ¿Deberíamos cambiar el algoritmo de alguna manera (por ejemplo, no ejecutar ningún proceso al mismo tiempo que la E / S)?

Mi solución tiene una campana y un silbato adicionales para asegurarme de que el orden de la salida sea el mismo que el de la entrada. Utilizo multiprocessing.queue's para enviar datos entre procesos, enviando mensajes de detención para que cada proceso sepa dejar de revisar las colas. Creo que los comentarios en la fuente deberían dejar en claro lo que está pasando, pero si no, házmelo saber.

#!/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:])

Llegar tarde a la fiesta ...

joblib tiene una capa encima del multiprocesamiento para ayudar a hacer bucles for paralelos. Le brinda facilidades como un despacho perezoso de trabajos y un mejor informe de errores además de su sintaxis muy simple.

Como descargo de responsabilidad, soy el autor original de joblib.

Me doy cuenta de que llego un poco tarde a la fiesta, pero recientemente descubrí el paralelo GNU y quiero mostrar lo fácil que es lograr esta tarea típica con él.

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

Algo como esto servirá para 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)

Paralelo se ejecutará sum.pypara cada línea input.csv(en paralelo, por supuesto), luego generará los resultados en sums. Claramente mejor que multiprocessingmolestias

Vieja escuela.

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

Aquí está la estructura final de multiprocesamiento.

python p1.py | python p2.py | python p3.py

Sí, el caparazón los ha unido a nivel de SO. Me parece más simple y funciona muy bien.

Sí, hay un poco más de sobrecarga al usar pickle (o cPickle). Sin embargo, la simplificación parece merecer el esfuerzo.

Si desea que el nombre del archivo sea un argumento p1.py, es un cambio fácil.

Más importante aún, una función como la siguiente es muy útil.

def get_stdin():
    while True:
        try:
            yield pickle.load( sys.stdin )
        except EOFError:
            return

Eso te permite hacer esto:

for item in get_stdin():
     process item

Esto es muy simple, pero no fácil le permiten tener múltiples copias de P2.py ejecución.

Tiene dos problemas: abanico y abanico. El P1.py debe expandirse de alguna manera a múltiples P2.py. Y los P2.py deben fusionar de alguna manera sus resultados en un solo P3.py.

El enfoque de la vieja escuela para la distribución en abanico es una arquitectura "Push", que es muy eficaz.

En teoría, la extracción de varios P2.py de una cola común es la asignación óptima de recursos. Esto suele ser ideal, pero también es una buena cantidad de programación. ¿Es realmente necesaria la programación? ¿O el procesamiento por turnos será lo suficientemente bueno?

En la práctica, encontrará que hacer que P1.py haga un simple trato "round robin" entre múltiples P2.py puede ser bastante bueno. Tendría P1.py configurado para hacer frente a n copias de P2.py a través de canalizaciones con nombre. Los P2.py leerían cada uno de su tubería correspondiente.

¿Qué pasa si un P2.py obtiene todos los datos del "peor caso" y se queda atrás? Sí, el round robin no es perfecto. Pero es mejor que solo un P2.py y puede abordar este sesgo con una simple aleatorización.

El abanico de múltiples P2.py a un P3.py es un poco más complejo, aún. En este punto, el enfoque de la vieja escuela deja de ser ventajoso. P3.py necesita leer desde múltiples conductos con nombre usando la selectbiblioteca para intercalar las lecturas.

Probablemente también sea posible introducir un poco de paralelismo en la parte 1. Probablemente no sea un problema con un formato que es tan simple como CSV, pero si el procesamiento de los datos de entrada es notablemente más lento que la lectura de los datos, podría leer fragmentos más grandes y luego continuar leyendo hasta encontrar un "separador de filas" ( newline en el caso de CSV, pero nuevamente eso depende del formato leído; no funciona si el formato es lo suficientemente complejo).

Estos fragmentos, cada uno de los cuales probablemente contenga varias entradas, se pueden transferir a una multitud de procesos paralelos que leen trabajos de una cola, donde se analizan y dividen, y luego se colocan en la cola para la etapa 2.