Resolviendo problemas paralelos vergonzosamente usando el multiprocesamiento de Python

¿Cómo se usa el multiprocesamiento para abordar problemas vergonzosamente paralelos ?

Los problemas paralelos de vergüenza consisten típicamente en tres partes básicas:

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

Podemos paralelizar el progtwig en dos dimensiones:

  • La Parte 2 puede ejecutarse en múltiples núcleos, ya que cada cálculo es independiente; El orden de procesamiento no importa.
  • Cada parte se puede ejecutar de forma independiente. La Parte 1 puede colocar los 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 los resultados de la cola de salida y escribirlos.

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

Aquí está el problema de ejemplo: dado un archivo CSV con filas de enteros como entrada, calcule sus sums. Separe el problema en tres partes, que pueden ejecutarse en paralelo:

  1. Procese el archivo de entrada en datos sin procesar (listas / iterables de enteros)
  2. Calcula las sums de los datos, en paralelo.
  3. Salida las sums

A continuación se encuentra el progtwig Python tradicional, con 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 progtwig y lo reescribamos para usar el multiprocesamiento para paralelizar las tres partes descritas anteriormente. A continuación se muestra un esqueleto de este nuevo progtwig paralelizado, que debe ser desarrollado 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:]) 

Estas piezas de código, así como otra pieza de código que puede generar archivos CSV de ejemplo para fines de prueba, se pueden encontrar en github .

Apreciaría cualquier idea sobre cómo abordarán este problema los gurús de la concurrencia.


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

  • ¿Debo tener procesos secundarios para leer los datos y colocarlos en la cola, o el proceso principal puede hacer esto sin bloquear hasta que se lea toda la entrada?
  • 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 sum?
    • En caso afirmativo, ¿a qué método invoco en la agrupación para que empiece a procesar los resultados que entran en 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, sino que podríamos esperar hasta que se analizaran todas las entradas y se calcularon todos los resultados (por ejemplo, porque sabemos que todas las entradas y salidas cabrán en la memoria del sistema). ¿Deberíamos cambiar el algoritmo de alguna manera (por ejemplo, no ejecutar ningún proceso simultáneamente con I / O)?

Mi solución tiene una campana y un silbato adicionales para garantizar que el orden de la salida sea el mismo que el de la entrada. Uso multiprocessing.queue’s para enviar datos entre procesos, enviando mensajes de parada para que cada proceso sepa dejar de revisar las colas. Creo que los comentarios en la fuente deberían aclarar lo que está sucediendo, pero si no me lo hacen 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:]) 

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

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

Algo como esto hará por 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 ejecutará sum.py para cada línea en input.csv (en paralelo, por supuesto), luego emitirá los resultados en sums . Claramente mejor que multiprocessing molestia

Llegando tarde a la fiesta …

joblib tiene una capa encima del multiprocesamiento para ayudar a hacer paralelos para los bucles. Le brinda servicios como el despacho lento de trabajos y un mejor informe de errores además de su muy simple syntax.

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

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 shell los ha unido a nivel del sistema operativo. Me parece más simple y funciona muy bien.

Sí, hay un poco más de sobrecarga en el uso de pickle (o cPickle). La simplificación, sin embargo, parece valer la pena.

Si desea que el nombre del archivo sea un argumento para 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 le permite tener múltiples copias de P2.py en ejecución.

Tienes dos problemas: fan-out y fan-in. El P1.py debe de alguna manera expandirse hacia varios P2.py. Y el P2.py de alguna manera debe combinar sus resultados en un solo P3.py.

El enfoque de la vieja escuela para fan-out es una architecture “Push”, que es muy efectiva.

Teóricamente, la extracción de múltiples recursos de P2.py de una cola común es la asignación óptima. Esto suele ser ideal, pero también es una buena cantidad de progtwigción. ¿Es realmente necesaria la progtwigción? ¿O será el procesamiento de round-robin lo suficientemente bueno?

En la práctica, encontrará que hacer que P1.py haga un simple “round robin” entre varios P2.py puede ser bastante bueno. Tendría P1.py configurado para tratar n copias de P2.py a través de canalizaciones con nombre. El P2.py cada uno leería de su tubo apropiado.

¿Qué pasa si una P2.py obtiene todos los datos del “peor caso” y se ejecuta muy por detrás? Sí, el round-robin no es perfecto. Pero es mejor que solo una P2.py y puede abordar este sesgo con una aleatorización simple.

El abanico de entradas de múltiples P2.py a una 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 de múltiples tuberías con nombre usando la biblioteca de select 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 sea 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, luego continuar leyendo hasta que encuentre un “separador de filas” ( nueva línea en el caso CSV, pero nuevamente, eso depende del formato leído; no funciona si el formato es lo suficientemente complejo).

Estos trozos, cada uno probablemente con múltiples entradas, se pueden agrupar en 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.