tipos - Solución de problemas vergonzosamente paralelos utilizando el multiproceso de Python

para que sirve el raise en python (5)

¿Cómo se usa el multiprocessing para abordar problemas embarazosamente paralelos ?

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

Lea datos de entrada (desde un archivo, base de datos, conexión tcp, etc.).
Ejecute cálculos en los datos de entrada, donde cada cálculo es independiente de cualquier otro cálculo .
Escriba 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 puede ejecutarse en múltiples 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 extraer datos de la cola de entrada y poner resultados en una cola de salida, y la Parte 3 puede extraer resultados de la cola de salida y escribirlos.

Este parece ser un patrón básico en la programación simultánea, pero todavía estoy perdido al tratar de resolverlo, así que vamos a escribir un ejemplo canónico para ilustrar cómo se hace con el multiprocesamiento .

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

Procesar el archivo de entrada en datos brutos (listas / iterables de enteros)
Calcule las sumas de los datos, en paralelo
Saca 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 reescribamos para usar multiprocesamiento para paralelizar las tres partes descritas arriba. A continuación se muestra un esqueleto de este nuevo programa paralelizado, que debe ser desarrollado para abordar las partes en los comentarios:

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 aquí sobre cómo sus gurús de concurrencia abordarían este problema.

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

¿Debo tener procesos hijo para leer en los datos y colocarlos en la cola, o puede el proceso principal hacer esto sin bloquear hasta que se lea toda la información?
Del mismo modo, ¿debería tener un proceso secundario para escribir los resultados desde la cola procesada, o puede el proceso principal hacer esto sin tener que esperar todos los resultados?
¿Debería usar un grupo de procesos para las operaciones de suma?
- En caso afirmativo, ¿a qué método llamo en el grupo para que comience a procesar los resultados que llegan 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 ingresamos, pero podríamos esperar hasta que se haya analizado toda la entrada y se hayan calculado todos los resultados (por ejemplo, porque sabemos que todas las entradas y salidas encajarán en la memoria del sistema). ¿Deberíamos cambiar el algoritmo de alguna manera (por ejemplo, no ejecutar ningún proceso simultáneamente con E / S)?

Llegando tarde a la fiesta ...

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

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

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

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)

Parallel ejecutará sum.py para cada línea en input.csv (en paralelo, por supuesto), luego input.csv los resultados a sums . Claramente mejor que problemas de multiprocessing

Mi solución tiene un toque extra para asegurar que el orden de la salida tenga el mismo orden de entrada. Utilizo multiprocessing.queue para enviar datos entre procesos, enviando mensajes de detención para que cada proceso sepa dejar de verificar las colas. Creo que los comentarios en la fuente deben dejar en claro lo que está sucediendo, pero si no me dejan 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:])

Probablemente sea posible introducir un poco de paralelismo en la parte 1 también. Probablemente no sea un problema con un formato 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" ( nueva línea en el caso de CSV, pero nuevamente eso depende del formato leído; no funciona si el formato es suficientemente complejo).

Estos fragmentos, que probablemente contengan varias 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 de la etapa 2.

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 en el nivel del sistema operativo. Me parece más simple y funciona muy bien.

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

Si desea que el nombre de 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 permite que se ejecuten varias copias de P2.py fácilmente.

Tienes dos problemas: fan-out y fan-in. El P1.py de alguna manera debe desplegarse en múltiples P2.py''s. Y los P2.py deben fusionar de algún modo sus resultados en un solo P3.py.

El enfoque de fan-out de la vieja escuela es una arquitectura "Push", que es muy efectiva.

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

Prácticamente, descubrirá que hacer que P1.py haga un simple "round robin" entre múltiples P2.py puede ser bastante bueno. Tendría P1.py configurado para tratar n copias de P2.py a través de conductos con nombre. Los P2.py leerían cada uno de ellos desde su tubería apropiada.

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

Fan-in de múltiples P2.py a uno 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 varias canalizaciones con nombre utilizando la biblioteca de select para intercalar las lecturas.