Was ist der beste Weg, um große Dateien in Python teilen für Multiprocessing?

Question

Ich laufe über viele „peinlichen parallel“ Projekte würde ich mit dem multiprocessing Modul parallelisieren mag. Doch oft sie beinhalten in großen Dateien (größer als 2 GB) zu lesen, verarbeiten sie Zeile für Zeile, laufen grundlegende Berechnungen und dann Ergebnisse zu schreiben. Was ist der beste Weg, um eine Datei zu teilen und es mit Python Multiprocessing-Modul zu verarbeiten? Sollte Queue oder JoinableQueue in multiprocessing verwendet werden? Oder die Queue selbst Modul? Oder sollte ich die Datei map iterable über einen Pool von Prozessen mit multiprocessing? Ich habe mit diesen Ansätzen experimentiert, aber der Aufwand ist immens in der Verteilung der Daten Zeile für Zeile. Ich habe durch die Verwendung cat file | process1 --out-file out1 --num-processes 2 | process2 --out-file out2 auf einem leichten Rohr Filter Design angesiedelt, die einen bestimmten Prozentsatz des ersten Prozesses der Eingang geht direkt an den zweiten Eingang (siehe dieser Beitrag ), aber ich mag eine Lösung haben, vollständig in Python enthalten ist.

Überraschenderweise ist die Python-Dokumentation keine kanonische Weise vorschlagen, dies zu tun (trotz einer langen Abschnitt über die Programmierrichtlinien in der multiprocessing Dokumentation).

Danke, Vince

Weitere Informationen: Die Bearbeitungszeit pro Zeile variiert. Einige Probleme sind schnell und kaum nicht-I / O gebunden, einige sind CPU-gebunden. Die CPU gebunden ist, nicht-abhängige Aufgaben der Post von Parallelisierung gewinnen, so dass auch ineffiziente Wege von Daten zu einer Verarbeitungsfunktion zuweisen noch in Bezug auf die Wanduhr Zeit von Vorteil wäre.

Ein gutes Beispiel ist ein Skript, das Felder aus Linien, Kontrollen für eine Vielzahl von bitweise Flags extrahiert und schreibt Zeilen mit bestimmten Flags in eine neue Datei in einem völlig neuen Format. Dies scheint wie ein I / O-gebundenes Problem, aber wenn ich lief es mit meiner billigen gleichzeitigen Ausführung mit Rohren, es war etwa 20% schneller. Wenn ich es mit Pool und Karte laufen, oder der Warteschlange in multiprocessing ist es immer über 100% langsamer.

Answer 1

Eines der besten Architekturen ist bereits Teil des Linux-Betriebssysteme. Keine besonderen Bibliotheken erforderlich.

Sie wollen ein "Fan-out" -Design.

1. A "main" Programm erzeugt eine Anzahl von Teilprozessen durch Rohre verbunden sind.

2. Das Hauptprogramm liest die Datei, Leitungen zu dem Pipes schreibt dabei die minimale Filterung erforderlich, um die Linien zu entsprechenden Teilprozessen zu behandeln.

Jede Subprozess soll wohl eine Pipeline von verschiedenen Prozessen, die von stdin lesen und schreiben.

Sie brauchen nicht eine Warteschlange Datenstruktur, das ist genau das, was eine In-Memory-Pipeline ist -. Eine Warteschlange von Bytes zwischen zwei gleichzeitigen Prozessen

Answer 2

Eine Strategie ist, damit jeder Arbeitnehmer einen Offset zuweisen wenn Sie also acht Arbeiter haben Prozesse, die Sie zuweisen dann Zahlen 0 bis 7 Worker Nummer 0 liest der erste Datensatz verarbeitet sie dann überspringt 7 und geht auf den 8. Aufzeichnung usw. zu verarbeiten, Arbeiter Nummer 1 liest der zweite Datensatz 7 dann überspringt und verarbeitet die neunte Rekord .........

Es gibt eine Reihe von Vorteilen auf dieses Schema. Es spielte keine Rolle, wie groß die Datei die Arbeit ist immer gleichmäßig aufgeteilt wird, verarbeitet auf der gleiche Maschine in etwa der gleichen Rate wird verarbeitet, und die gleichen Pufferbereiche verwenden, so dass Sie ein übermäßiges E / A-Overhead entstehen nicht. Solange die hasnt Datei aktualisiert wurden, können Sie einzelne Threads erneut ausführen, aus Fehlern zu erholen.

Answer 3

Sie nicht erwähnen, wie Sie die Zeilen verarbeiten; möglicherweise der wichtigste Teil der Informationen.

Ist jede unabhängige Linie? Ist die Berechnung auf einer Linie abhängig vor dem nächsten kommt? Müssen sie in Blöcken verarbeitet werden? Wie lange dauert die Verarbeitung für jede Zeile nehmen? Gibt es einen Verarbeitungsschritt, der „alle“, um die Daten am Ende übernehmen müssen? Oder können Zwischenergebnisse weggeworfen werden und nur eine laufende Summe beibehalten? Kann die Datei durch Teilen Größe durch Zählung der Fäden zunächst geteilt? Oder macht es wachsen, wie Sie es verarbeiten?

Wenn die Linien sind unabhängige und die Datei nicht wächst, ist die einzige Koordination Sie brauchen, ist „Startadressen“ auf dem Bauernhof und „Längen“ zu jedem der Arbeiter; sie können unabhängig voneinander geöffnet und in die Datei suchen und dann müssen Sie einfach ihre Ergebnisse koordinieren; vielleicht für N Ergebnisse warten in eine Warteschlange zu kommen.

Wenn die Linien nicht unabhängig sind, wird die Antwort hängt stark von der Struktur der Datei.

Answer 4

Ich weiß, dass Sie speziell über Python gefragt, aber ich werde empfehlen Ihnen, bei Hadoop suchen ( http: // hadoop. apache.org/ ): es implementiert die Map und Reduce-Algorithmus, der speziell diese Art von Problem zu lösen wurde entwickelt.

Viel Glück

Answer 5

Es hängt viel von dem Format der Datei.

Ist es sinnvoll, es überall zu spalten? Oder benötigen Sie es auf einer neuen Zeile zu teilen? Oder müssen Sie sicherstellen, dass Sie es am Ende einer Objektdefinition geteilt?

Statt die Datei zu spalten, sollten Sie mehrere Leser auf der gleichen Datei verwenden, mit os.lseek an den entsprechenden Teil der Datei zu springen.

Update: Poster fügte hinzu, dass er sich auf neue Linien spalten will. Dann schlage ich folgende Möglichkeiten:

Angenommen, Sie 4 Prozesse haben. Dann ist die einfache Lösung ist auf 0%, 25%, 50% und 75% der Datei os.lseek und Bytes lesen, bis Sie die erste neue Linie getroffen. Das ist Ihr Ausgangspunkt für jeden Prozess. Sie brauchen nicht die Datei zu teilen, dies zu tun, nur in jedem Prozess an die richtige Stelle in der großen Datei suchen und starten von dort zu lesen.

Answer 6

Fredrik Lundh Einige Hinweise zu Tim Bray Wide Finder Benchmark eine interessante Lektüre ist, über einen sehr ähnlichen Anwendungsfall, mit vielen guten Ratschlägen. Verschiedene andere Autoren implementiert auch die gleiche Sache, einige aus dem Artikel verknüpft sind, aber Sie könnten für „Python breit Finder“ oder etwas googeln versuchen wollen etwas mehr zu finden. (Es gab auch eine Lösung irgendwo auf der Grundlage des multiprocessing Modul, aber das scheint nicht mehr vorhanden zu sein)

Answer 7

Wenn die Laufzeit lang ist, statt jeden Prozess mit las die nächste Zeile durch einen Queue, müssen die Prozesse Chargen von Zeilen zu lesen. Auf diese Weise der Aufwand wird über mehrere Zeilen abgeschrieben (z Tausende oder mehr).