Lösen Sie peinlich parallele Probleme mit Python Multiprocessing

Wie nutzt man Multiprocessing , um peinlich parallele Probleme anzugehen ?

Peinlich parallele Probleme bestehen normalerweise aus drei Grundteilen:

1. Lesen Sie Eingabedaten (aus einer Datei, Datenbank, TCP-Verbindung usw.).
2. Führen Sie Berechnungen für die Eingabedaten aus, wobei jede Berechnung von jeder anderen Berechnung unabhängig ist .
3. Schreiben Sie Ergebnisse von Berechnungen (in eine Datei, Datenbank, 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. Reihenfolge der Verarbeitung spielt keine Rolle.
• Jeder Teil kann unabhängig 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 grundlegendstes Muster bei der gleichzeitigen Programmierung zu sein, aber ich bin immer noch nicht in dem Versuch, es zu lösen. Schreiben wir also ein kanonisches Beispiel, um zu veranschaulichen, wie dies mithilfe von Multiprocessing gemacht wird .

Hier ist das Beispielproblem: Berechnen Sie bei einer CSV-Datei mit Zeilen von Ganzzahlen als Eingabe deren Summen. Teilen Sie das Problem in drei Teile auf, die alle parallel ablaufen können:

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

Im Folgenden finden Sie ein traditionelles, an einen Prozess gebundenes 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 mithilfe der Mehrfachverarbeitung die drei oben beschriebenen Teile zu parallelisieren. Unten finden Sie ein Grundgerüst dieses neuen, parallelisierten 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 weiterer Codeteil, der zu Testzwecken CSV-Beispieldateien generieren kann, befinden sich auf github .

Ich würde mich über jeden Einblick freuen, wie Sie Parallelitätsgurus dieses Problem angehen würden.

Hier sind einige Fragen, die ich hatte, als ich über dieses Problem nachdachte. Bonuspunkte für die Adressierung aller / aller:

• Sollte ich untergeordnete Prozesse zum Einlesen und Einfügen der Daten in die Warteschlange haben, oder kann der Hauptprozess dies tun, ohne zu blockieren, bis alle Eingaben gelesen wurden?
• Sollte ich einen untergeordneten Prozess zum Ausschreiben 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?
  • Wenn ja, welche Methode rufe ich für den Pool auf, damit dieser die in die Eingabewarteschlange eingehenden Ergebnisse verarbeitet, ohne auch die Eingabe- und Ausgabeprozesse zu blockieren? apply_async () ? map_async () ? imap () ? imap_unordered () ?
• Angenommen, wir müssten die Eingabe- und Ausgabewarteschlangen nicht während der Dateneingabe entfernen, sondern könnten warten, bis alle Eingaben analysiert und alle Ergebnisse berechnet wurden (z. B. weil wir wissen, dass alle Eingaben und Ausgaben in den Systemspeicher passen). Sollten wir den Algorithmus in irgendeiner Weise ändern (z. B. keine Prozesse gleichzeitig mit E / A ausführen)?

Meine Lösung verfügt über eine zusätzliche Glocke und Pfeife, 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 klar machen, 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:])

Komme spät zur Party ...

joblib verfügt über eine Ebene über der Mehrfachverarbeitung , um die Parallelisierung für Schleifen zu erleichtern . Es bietet Ihnen Funktionen wie ein verzögertes Versenden von Jobs und eine bessere Fehlerberichterstattung zusätzlich zu seiner sehr einfachen Syntax.

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

Mir ist klar, dass ich etwas spät zur Party komme, aber ich habe kürzlich GNU parallel entdeckt und möchte zeigen, wie einfach es ist, diese typische Aufgabe damit zu erledigen.

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

So etwas reicht für 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 wird sum.pyfür jede Zeile in input.csv(natürlich parallel) ausgeführt und gibt dann die Ergebnisse an aus sums. Deutlich besser als multiprocessingÄrger

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 zusammengestrickt. Es scheint mir einfacher und es funktioniert sehr gut.

Ja, die Verwendung von pickle (oder cPickle) ist etwas aufwändiger. Die Vereinfachung scheint jedoch die Mühe wert zu sein.

Wenn Sie möchten, dass der Dateiname ein Argument für ist p1.py ist, 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

So können Sie Folgendes tun:

for item in get_stdin():
     process item

Das ist sehr einfach, aber nicht einfach ermöglicht es Ihnen , mehrere Kopien von P2.py am Laufen zu haben.

Sie haben zwei Probleme: Fan-Out und Fan-In. Der P1.py muss sich irgendwie auf mehrere P2.py auffächern. Und die P2.py müssen ihre Ergebnisse irgendwie zu einem einzigen P3.py zusammenführen.

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

Theoretisch ist das Abrufen mehrerer P2.py 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 ist die Round-Robin-Verarbeitung gut genug?

In der Praxis werden Sie feststellen, dass es ziemlich gut sein kann, P1.py zu einem einfachen "Round Robin" zu machen, das sich mit mehreren P2.py befasst. Sie hätten 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 zurückliegt? Ja, Round-Robin ist nicht perfekt. Aber es ist besser als nur ein P2.py und Sie können diese Verzerrung durch einfache Randomisierung beheben.

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

Es ist wahrscheinlich auch möglich, ein bisschen Parallelität in Teil 1 einzuführen. Wahrscheinlich kein Problem mit einem Format, das so einfach wie CSV ist, aber wenn die Verarbeitung der Eingabedaten merklich langsamer ist als das Lesen der Daten, können Sie größere Blöcke lesen und dann weiterlesen, bis Sie ein "Zeilentrennzeichen" finden ( Zeilenumbruch im CSV-Fall, aber auch dies hängt 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 Menge paralleler Prozesse weitergeleitet werden, die Jobs aus einer Warteschlange lesen, wo sie analysiert und aufgeteilt werden, und dann für Stufe 2 in die Warteschlange gestellt werden.