die Tasche nicht alle Kerne verwenden? Alternativen?

Question

Ich habe ein Python-Skript, das Folgendes tut: i. die eine Eingabedatei mit Daten verwendet (normalerweise geschachteltes JSON-Format) ii. übergibt die Daten Zeile für Zeile an eine andere Funktion, die die Daten in das gewünschte Format manipuliert. iii. und schließlich schreibt es die Ausgabe in eine Datei.

Hier ist meine aktuelle einfache Python Linie, das dies tut ...

def manipulate(line): 
    # a pure python function which transforms the data 
    # ... 
    return manipulated_json 

for line in f: 
    components.append(manipulate(ujson.loads(line))) 
    write_to_csv(components)` 

Dies funktioniert, aber mit dem Python GIL es zu einem Kern auf dem Server zu begrenzen, ist es sehr langsam, vor allem mit großen Mengen an Daten.

Die Menge der Daten, mit denen ich normalerweise zu tun habe, ist etwa 4 Gigs gzip komprimiert, aber gelegentlich muss ich Daten verarbeiten, die Hunderte von Gigs gzip komprimiert ist. Es ist nicht unbedingt Big Data, kann aber immer noch nicht alle im Speicher verarbeitet werden und mit Pythons GIL ist sehr langsam zu verarbeiten.

Auf der Suche nach einer Lösung zur Optimierung unserer Datenverarbeitung stieß ich auf dask. Während PySpark zu dieser Zeit die offensichtliche Lösung für mich zu sein schien, überzeugten mich die Versprechen von dask und seine Einfachheit und ich beschloss, es zu versuchen.

Nach viel Forschung in dask und wie man es benutzt, habe ich ein sehr kleines Skript zusammengestellt, um meinen aktuellen Prozess zu replizieren. Das Skript sieht wie folgt aus:

import dask.bag as bag 
import json 
bag.from_filenames('input.json.gz').map(json.loads).map(lambda x:manipulate(x)).concat().to_dataframe().to_csv('output.csv.gz')` 

Dies funktioniert und erzeugt die gleichen Ergebnisse wie meine ursprüngliche nicht-dask Skript, aber es noch verwendet nur eine CPU auf dem Server. Also hat es überhaupt nicht geholfen. In der Tat ist es langsamer.

Was mache ich falsch? Fehle ich etwas? Ich bin noch ziemlich neu zu dask, lass es mich wissen, wenn ich etwas übersehen habe oder wenn ich etwas ganz anderes machen sollte.

Gibt es auch Alternativen zu DASK für die Nutzung der vollen Kapazität des Servers (d. H. Alle CPUs) für was ich tun muss?

Danke,

T

Answer 1

Das Problem hier ist mit dask.dataframe.to_csv, die Sie auf Single-Core-Modus zwingt.

Ich empfehle, dask.bag zu verwenden, um Ihre Lese- und Bearbeitungsvorgänge durchzuführen und dann auf eine Reihe von CSV-Dateien parallel abzulegen. Das Dumping zu vielen CSV-Dateien ist viel einfacher zu koordinieren als das Dumping zu einer einzelnen CSV-Datei.

import dask.bag as bag 
import json 
b = bag.from_filenames('input.json.gz').map(json.loads).map(manipulate).concat() 
b.map(lambda t: ','.join(map(str, t)).to_textfiles('out.*.csv').compute() 

Es könnte auch ein Problem mit dem Versuch, eine einzelne GZIP Datei parallel zu lesen, aber die oben genannten sollten Sie loslegen.

Answer 2

Es scheint, dass Taschen nur so parallel sind wie die Anzahl der Partitionen, die sie haben.

Für mich ergibt

mybag=bag.from_filenames(filename, chunkbytes=1e7) 
mybag.npartitions 

laufen

, die das Problem gelöst und machte die Verarbeitung vollständig parallelizable.

Answer 3

Wenn Sie einen Glob-basierten Dateinamen angeben, z. MyFiles-*.csv zu demask dataframe.to_csv() sollten Sie in der Lage sein, den Datenrahmen auf Festplatte auszugeben. Es erstellt mehrere Dateien anstelle von 1 großen CSV-Datei. Siehe diesen Thread für mre https://groups.google.com/a/continuum.io/forum/#!searchin/blaze-dev/to_csv/blaze-dev/NCQfCoOWEcI/S7fwuCfeCgAJ

MyFiles-0001.csv 
MyFiles-0002.csv 
....