Résoudre des problèmes parallèles embarrassants en utilisant le multitraitement Python

Question

Comment utiliser multiprocessing pour s'attaquer à embarrassingly parallel problems?

problèmes embarassingly parallèles sont généralement constitués de trois parties:

1. Lire les données d'entrée (à partir d'un fichier, la base de données, la connexion TCP, etc.).
2. Exécuter calculs sur les données d'entrée, où chaque calcul est indépendant de tout autre calcul.
3. Ecrivez résultats de calculs (dans un fichier, une base de données, une connexion tcp, etc.).

Nous pouvons paralléliser le programme en deux dimensions:

• Partie 2 peut fonctionner sur plusieurs cœurs, puisque chaque calcul est indépendant; l'ordre de traitement n'a pas d'importance.
• Chaque pièce peut s'exécuter indépendamment. La partie 1 peut placer des données dans une file d'attente d'entrée, la partie 2 peut extraire des données de la file d'entrée et placer les résultats dans une file d'attente de sortie et la partie 3 peut extraire les résultats de la file d'attente.

Cela semble un modèle le plus élémentaire dans la programmation concurrente, mais je suis toujours perdu en essayant de le résoudre, alors nous allons écrire un exemple canonique pour illustrer comment cela se fait à l'aide multitraitement.

Voici l'exemple de problème: Étant donné un CSV file avec des lignes d'entiers en entrée, calculez leurs sommes. Séparer le problème en trois parties, qui peuvent tous fonctionner en parallèle:

1. processus, le fichier d'entrée en des données brutes (listes/iterables d'entiers)
2. calculer les sommes des données, en parallèle
3. sortie les sommes

Ci-dessous est traditionnelle, un seul processus programme Python lié qui résout ces trois tâches:

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

Prenons ce programme et réécrivons-le pour utiliser le multitraitement pour paralléliser les trois parties décrites ci-dessus. Ci-dessous un squelette de ce nouveau programme parallélisé, qui doit être étoffée pour répondre aux parties dans les commentaires:

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

Ces morceaux de code, ainsi que another piece of code that can generate example CSV files à des fins de test, peuvent être found on github.

J'apprécierais n'importe quelle perspicacité ici quant à la façon dont vous concurrenciez des gourous approcheriez ce problème.

---

Voici quelques questions que j'avais en pensant à ce problème. points de bonus pour adresser toute/tous:

• Dois-je avoir des processus enfants pour la lecture des données et en le plaçant dans la file d'attente, ou puis le processus principal le faire sans bloquer jusqu'à ce que toute entrée est lue?
• De même, devrais-je avoir un processus fils pour écrire les résultats de la file d'attente traitée, ou est-ce que le processus principal peut le faire sans avoir à attendre tous les résultats?
• Dois-je utiliser un processes pool pour les opérations de somme? Si oui, quelle méthode dois-je appeler sur le pool pour qu'il commence à traiter les résultats entrant dans la file d'attente d'entrée, sans bloquer les processus d'entrée et de sortie, aussi? apply_async()? map_async()? imap()? imap_unordered()? Supposons que nous n'ayons pas besoin de siphonner les files d'entrée et de sortie lorsque les données sont saisies, mais que nous pouvons attendre que toutes les entrées soient analysées et que tous les résultats soient calculés (par exemple, nous connaissons toutes les entrées et sorties). s'intégrer dans la mémoire du système). Devrions-nous modifier l'algorithme de quelque façon que ce soit (par exemple, ne pas exécuter de processus en même temps que les E/S)?

Answer 1

Ma solution a une cloche et un sifflet supplémentaire pour s'assurer que l'ordre de la sortie a le même que l'ordre de l'entrée. J'utilise multiprocessing.queue pour envoyer des données entre les processus, en envoyant des messages d'arrêt afin que chaque processus sache arrêter de vérifier les files d'attente. Je pense que les commentaires dans la source devraient indiquer clairement ce qui se passe, mais si ce n'est pas le cas, faites-le moi savoir.

#!/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

Old School.

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 

est ici la structure finale multi-traitement.

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

Oui, le shell les a tricotés ensemble au niveau de l'OS. Cela me semble plus simple et cela fonctionne très bien.

Oui, il y a un peu plus de frais généraux dans l'utilisation du pickle (ou cPickle). La simplification semble cependant valoir l'effort.

Si vous voulez que le nom de fichier soit un argument à p1.py, c'est un changement facile. Plus important encore, une fonction comme celle-ci est très pratique.

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

qui vous permet de le faire:

for item in get_stdin(): 
    process item 

---

Ceci est très simple, mais il ne facilement vous permettre d'avoir plusieurs copies de P2.py en cours d'exécution.

Vous avez deux problèmes: le ventilateur et le ventilateur. Le P1.py doit en quelque sorte se déployer sur plusieurs P2.py. Et les P2.py doivent en quelque sorte fusionner leurs résultats en un seul P3.py.

L'approche old-school du fan-out est une architecture «Push», très efficace.

Théoriquement, l'extraction multiple de P2.py à partir d'une file d'attente commune est l'allocation optimale des ressources. C'est souvent idéal, mais c'est aussi une bonne quantité de programmation. La programmation est-elle vraiment nécessaire? Ou le traitement de la ronde sera-t-il suffisant?

Pratiquement, vous trouverez que faire P1.py faire un simple "round robin" traitant de plusieurs P2.py peut être assez bon. Vous devez avoir P1.py configuré pour traiter n copies de P2.py via des canaux nommés. Les P2.py liraient chacun leur tube approprié.

Que se passe-t-il si un fichier P2.py obtient toutes les données du "pire cas" et qu'il est très en retard? Oui, round-robin n'est pas parfait. Mais c'est mieux qu'une seule P2.py et vous pouvez corriger ce biais avec une simple randomisation.

Fan-in de plusieurs P2.py à un P3.py est encore un peu plus complexe. À ce stade, l'approche de la vieille école cesse d'être avantageuse. P3.py doit lire à partir de plusieurs canaux nommés en utilisant la bibliothèque select pour entrelacer les lectures.

Answer 3

Il est probablement possible d'introduire également un peu de parallélisme dans la partie 1. Probablement pas un problème avec un format aussi simple que CSV, mais si le traitement des données d'entrée est sensiblement plus lent que la lecture des données, vous pouvez lire des morceaux plus gros, puis continuer à lire jusqu'à trouver un "séparateur de ligne" (newline dans le cas CSV, mais encore une fois cela dépend du format lu, ne fonctionne pas si le format est suffisamment complexe).

Ces blocs, contenant probablement plusieurs entrées, peuvent ensuite être exportés vers une foule de processus parallèles lisant les travaux d'une file d'attente, où ils sont analysés et divisés puis placés dans la file d'attente pour l'étape 2.

Answer 4

Je me rends compte que je suis un peu en retard pour la fête, mais j'ai récemment découvert GNU parallel, et je veux montrer à quel point il est facile d'accomplir cette tâche typique avec.

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

Quelque chose comme ça pour faire 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) 

parallèle se déroulera sum.py pour chaque ligne input.csv (en parallèle, bien sûr), puis sortie les résultats sums. Il est clair que mieux que multiprocessing tracas

Answer 5

arrivé en retard à la fête ...

joblib a une couche sur le dessus de multitraitement pour aider à rendre parallèles pour les boucles. Il vous donne des facilités comme un envoi paresseux de tâches, et de meilleurs rapports d'erreurs en plus de sa syntaxe très simple. En tant que déni de responsabilité, je suis l'auteur original de joblib.