5 Frage: Lösen von peinlich parallelen Problemen mit Python Multiprocessing

Frage erstellt am Thu, Mar 11, 2010 12:00 AM

Wie verwendet man Multiprocessing , um peinlich parallele Probleme ?

Peinlich parallele Probleme bestehen normalerweise aus drei Hauptteilen:

  1. Eingabedaten lesen (aus einer Datei, Datenbank, TCP-Verbindung usw.).
  2. Führen Sie Berechnungen für die Eingabedaten durch, wobei jede Berechnung unabhängig von jeder anderen Berechnung ist.
  3. Schreibe Ergebnisse von Berechnungen (in eine Datei, eine Datenbank, eine TCP-Verbindung usw.)

Wir können das Programm in zwei Dimensionen parallelisieren:

  • Teil 2 kann auf mehreren Kernen ausgeführt werden, da jede Berechnung unabhängig ist. Die Reihenfolge der Verarbeitung spielt keine Rolle.
  • Jeder Teil kann unabhängig voneinander ausgeführt werden. Teil 1 kann Daten in eine Eingabewarteschlange stellen, Teil 2 kann Daten aus der Eingabewarteschlange abrufen und Ergebnisse in eine Ausgabewarteschlange stellen, und Teil 3 kann Ergebnisse aus der Ausgabewarteschlange abrufen und ausschreiben.

Dies scheint ein grundlegendes Muster bei der gleichzeitigen Programmierung zu sein, aber ich bin immer noch nicht in der Lage, es zu lösen. Lassen Sie uns ein kanonisches Beispiel schreiben, um zu veranschaulichen, wie dies mithilfe von Mehrfachverarbeitung erfolgt.

Hier ist das Beispielproblem: Geben Sie eine CSV-Datei mit Ganzzahlzeilen als Eingabe ein berechnen ihre Summen. Teilen Sie das Problem in drei Teile auf, die alle parallel ausgeführt werden können:

  1. Verarbeiten Sie die Eingabedatei in Rohdaten (Listen /Iterablen von Ganzzahlen).
  2. Berechnen Sie die Summen der Daten parallel
  3. Geben Sie die Summen aus

Nachfolgend finden Sie ein traditionelles, prozessgebundenes Python-Programm, das diese drei Aufgaben löst:

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

Nehmen wir dieses Programm und schreiben es neu, um die drei oben beschriebenen Teile mithilfe von Multiprocessing zu parallelisieren. Im Folgenden finden Sie ein Skelett dieses neuen, parallelen Programms, das ausgearbeitet werden muss, um die Teile in den Kommentaren zu behandeln:

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

Diese Codeteile sowie ein weiteres Codeteil, das ein Beispiel generieren kann CSV-Dateien können zu Testzwecken auf github gefunden werden

Ich würde mich über einen Einblick darüber freuen, wie Sie als Nebenläufige Gurus dieses Problem angehen würden.


Hier sind einige Fragen, die ich beim Nachdenken über dieses Problem hatte. Bonuspunkte für die Bearbeitung eines /aller:

  • Sollte ich untergeordnete Prozesse zum Einlesen und Einreihen der Daten in die Warteschlange haben, oder kann der Hauptprozess dies tun, ohne zu blockieren, bis alle Eingaben gelesen wurden?
  • Sollte ich auch einen untergeordneten Prozess zum Schreiben der Ergebnisse aus der verarbeiteten Warteschlange haben, oder kann der Hauptprozess dies tun, ohne auf alle Ergebnisse warten zu müssen?
  • Soll ich einen Prozesspool für die Summenoperationen verwenden? ?
  • Angenommen, wir mussten die Eingabe- und Ausgabewarteschlangen nicht während der Dateneingabe löschen, sondern konnten warten, bis alle Eingaben analysiert und alle Ergebnisse berechnet wurden (z. B. weil wir wissen, dass alle Eingaben und Ausgaben in das System passen Erinnerung). Sollten wir den Algorithmus in irgendeiner Weise ändern (z. B. keine Prozesse gleichzeitig mit E /A ausführen)?
78
  1. Haha, ich liebe den Begriff peinlich-parallel. Ich bin überrascht, dass dies das erste Mal ist, dass ich den Begriff höre. Es ist eine großartige Möglichkeit, sich auf dieses Konzept zu beziehen.
    2010-03-11 17: 28: 30Z
5 Antworten                              5                         

Meine Lösung verfügt über einen zusätzlichen Schnickschnack, um sicherzustellen, dass die Reihenfolge der Ausgabe mit der Reihenfolge der Eingabe übereinstimmt. Ich verwende multiprocessing.queue's, um Daten zwischen Prozessen zu senden und Stoppnachrichten zu senden, damit jeder Prozess die Überprüfung der Warteschlangen beenden kann. Ich denke, die Kommentare in der Quelle sollten klarstellen, was los ist, aber wenn nicht, lassen Sie es mich wissen.

 
#!/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:])
    
65
2015-03-03 22: 06: 44Z
  1. Dies ist die einzige Antwort, die tatsächlich multiprocessing verwendet hat. Das Kopfgeld geht an Sie, Sir.
    2010-03-12 16: 59: 41Z
  2. Ist es tatsächlich erforderlich, join für die Eingabe- und Zahlenverarbeitung aufzurufen? Könnten Sie nicht davonkommen, sich nur dem Ausgabeprozess anzuschließen und die anderen zu ignorieren? Wenn ja, gibt es noch einen guten Grund, join für alle anderen Prozesse aufzurufen?
    2011-12-04 03: 41: 43Z
  3. Schöne Antwort, muss ich sagen. Auf jeden Fall für immer mit einem Lesezeichen versehen!
    2012-03-09 05: 19: 29Z
  4. "Threads können" - "Daten zwischen Threads senden" - Threads und Prozesse werden beendet sehr verschieden. Ich sehe, dass dies für Anfänger verwirrend sein kann. Umso wichtiger ist es, die richtige Terminologie in einer Antwort zu verwenden, die so oft aufgegriffen wurde. Sie starten hier neue Prozesse. Sie spawnen im aktuellen Prozess nicht nur Threads.
    2015-03-02 17: 42: 14Z
  5. Fair genug. Ich habe den Text korrigiert.
    2015-03-03 22: 07: 03Z

Mir ist klar, dass ich für die Party etwas spät dran bin, aber ich habe kürzlich festgestellt, dass GNU parallel läuft und möchten zeigen, wie einfach es ist, diese typische Aufgabe damit zu lösen.

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

So etwas ist für sum.py ausreichend:

 
#!/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 führt sum.py für jede Zeile in input.csv aus (natürlich parallel) und gibt dann die Ergebnisse an sums aus. Deutlich besser als multiprocessing Ärger     

5
2013-08-23 11: 00: 50Z
  1. Parallele GNU-Dokumente rufen für jede Zeile in der Eingabedatei einen neuen Python-Interpreter auf. Der Mehraufwand beim Starten eines neuen Python-Interpreters (etwa 30 Millisekunden für Python 2.7 und 40 Millisekunden für Python 3.3 auf meinem i7 MacBook Pro mit einem Solid-State-Laufwerk) kann die Zeit, die zum Verarbeiten einer einzelnen Datenzeile erforderlich ist, erheblich übersteigen und zu einem Fehler führen viel verschwendete Zeit und schlechtere Gewinne als erwartet. Im Falle Ihres Beispielproblems würde ich wahrscheinlich nach Multiprocessing greifen .Pool .
    2013-08-23 16: 02: 31Z

Komm zu spät zur Party ...

Die

Joblib verfügt über eine Ebene über Multiprocessing, um die Parallelisierung von Schleifen zu erleichtern. Es bietet Ihnen Funktionen wie eine verzögerte Verteilung von Aufträgen und eine bessere Fehlerberichterstattung zusätzlich zu seiner sehr einfachen Syntax.

Als Haftungsausschluss bin ich der ursprüngliche Autor von joblib.

    
5
2014-01-15 08: 13: 10Z
  1. Ist Joblib also in der Lage, die E /A parallel zu verarbeiten, oder müssen Sie dies manuell tun? Können Sie ein Codebeispiel mit Joblib bereitstellen? Vielen Dank!
    2017-08-08 10: 31: 53Z

Alte Schule.

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

Hier ist die endgültige Struktur für die Mehrfachverarbeitung.

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

Ja, die Shell hat diese auf Betriebssystemebene zusammengefügt. Es scheint mir einfacher und es funktioniert sehr gut.

Ja, mit pickle (oder cPickle) ist etwas mehr Aufwand verbunden. Die Vereinfachung scheint jedoch die Mühe wert zu sein.

Wenn der Dateiname ein Argument für p1.py sein soll, ist dies eine einfache Änderung.

Noch wichtiger ist, dass eine Funktion wie die folgende sehr praktisch ist.

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

Damit können Sie Folgendes tun:

 
for item in get_stdin():
     process item

Dies ist sehr einfach, aber es ist nicht einfach , mehrere Kopien von P2.py auszuführen.

Sie haben zwei Probleme: Fan-Out und Fan-In. Das P1.py muss sich irgendwie auf mehrere P2.pys aufteilen. Und die P2.pys müssen irgendwie ihre Ergebnisse zu einer einzigen P3.py zusammenführen.

Der Old-School-Ansatz für Fan-Out ist eine "Push" -Architektur, die sehr effektiv ist.

Theoretisch ist das Abrufen mehrerer P2.pys aus einer gemeinsamen Warteschlange die optimale Zuweisung von Ressourcen. Dies ist oft ideal, aber es ist auch eine Menge Programmierung. Ist die Programmierung wirklich notwendig? Oder reicht die Round-Robin-Verarbeitung aus?

In der Praxis werden Sie feststellen, dass es ziemlich gut sein kann, P1.py dazu zu bringen, ein einfaches "Round-Robin" -Verfahren mit mehreren P2.pys durchzuführen. Sie haben P1.py so konfiguriert, dass n Kopien von P2.py über Named Pipes verarbeitet werden. Die P2.py's würden jeweils aus ihrer entsprechenden Pipe lesen.

Was ist, wenn ein P2.py alle "Worst-Case" -Daten erhält und weit hinterherläuft? Ja, Round-Robin ist nicht perfekt. Aber es ist besser als nur eine P2.py und Sie können diese Tendenz durch einfache Randomisierung beheben.

Das Fan-In von mehreren P2.pys zu einem P3.py ist noch etwas komplexer. An diesem Punkt ist der Old-School-Ansatz nicht mehr vorteilhaft. P3.py muss mithilfe der select-Bibliothek aus mehreren Named Pipes lesen, um die Lesevorgänge zu verschachteln.

    
4
01.03.2010 22: 23: 18Z
  1. Wäre dies nicht haariger, wenn ich n-Instanzen von p2.py starten möchte? Lassen Sie diese m-Chunks von r-Zeilen, die von p1.py ausgegeben werden, und verarbeiten Soll p3.py die m x r-Ergebnisse aus allen n p2.py-Instanzen abrufen?
    01.03.2010 22: 06: 49Z
  2. Ich habe diese Anforderung in der Frage nicht gesehen. (Vielleicht war die Frage zu lang und zu komplex, um diese Anforderung hervorzuheben.) Wichtig ist, dass Sie einen guten Grund haben sollten, damit zu rechnen, dass mehrere P2 Ihr Leistungsproblem tatsächlich lösen. Wir können zwar die Hypothese aufstellen, dass eine solche Situation besteht, aber die * nix-Architektur hatte dies noch nie und niemand hat es für angebracht gehalten, sie hinzuzufügen. Es kann hilfreich sein, mehrere P2s zu haben. Aber in den letzten 40 Jahren hat niemand genug gesehen, um es zu einem erstklassigen Teil der Shell zu machen.
    01.03.2010 22: 14: 10Z
  3. Das ist also meine Schuld. Lassen Sie mich diesen Punkt bearbeiten und klarstellen. Kommt die Verwirrung von der Verwendung von sum(), um mir zu helfen, die Frage zu verbessern? Dies dient nur zur Veranschaulichung. Ich hätte es durch do_something() ersetzen können, aber ich wollte ein konkretes, leicht verständliches Beispiel (siehe ersten Satz). In Wirklichkeit ist mein do_something() sehr CPU-intensiv, aber peinlich parallelisierbar, da jeder Anruf unabhängig ist. Daher helfen mehrere Kerne, die darauf kauen.
    01.03.2010 22: 38: 04Z
  4. "Kommt die Verwirrung von der Verwendung von sum ()?" Ganz sicher nicht. Ich bin mir nicht sicher, warum Sie es erwähnen würden. Sie sagten: "Wäre das nicht haariger, wenn ich n Instanzen von p2.py starten möchte?" Ich habe diese Anforderung in der Frage nicht gesehen.
    01.03.2010 23: 28: 28Z

Es ist wahrscheinlich möglich, auch in Teil 1 ein wenig Parallelität einzuführen. Wahrscheinlich kein Problem mit einem Format, das so einfach wie CSV ist, aber wenn die Verarbeitung der Eingabedaten merklich langsamer als das Lesen der Daten ist, können Sie größere Abschnitte lesen und dann weiterlesen, bis Sie ein "Zeilentrennzeichen" finden ( Zeilenumbruch im CSV-Fall, dies hängt jedoch wiederum vom gelesenen Format ab; funktioniert nicht, wenn das Format ausreichend komplex ist).

Diese Blöcke, von denen jeder wahrscheinlich mehrere Einträge enthält, können dann an eine Vielzahl von parallelen Prozessen weitergeleitet werden, die Jobs aus einer Warteschlange lesen, wo sie analysiert und aufgeteilt und dann für Stufe 2 in die Warteschlange gestellt werden.

p>     
0
2010-03-10 16: 39: 09Z
Quelle platziert Hier