DASK Dataframe in Schleifen

Question

Ich habe ein paar Probleme mit dem Versuch, Schleifen in Dask zu implementieren. zum Beispiel in dem folgenden Code:

for i in range(len(col)): 
    if df[col[i]].dtype=='object': 
     pass 
    elif df[col[i]].std().compute()==0: 
     cols_constant.append(col[i]) 
df = df.drop(cols_constant,axis=1) 

Der gleiche Code ist sehr schnell Pandas verwenden, aber auf dask ist es eine beträchtliche Menge an Zeit zu nehmen, um die Aufgabe abzuschließen.

Ich verstehe Dask ist ineffiziente Überschleifen. Aber wie kann ich meinen Code für Dask für ähnliche Funktionen optimieren?

Ich kann e.persist() nicht verwenden, da wir beabsichtigen, die Berechnung auf mehreren Worker-Systemen durchzuführen.

Wird es nützlich sein, die Funktion 'dask.do' zu verwenden, um die gleiche Aufgabe zu parallelisieren?

Answer 1

Jedes Mal, wenn Sie df.column.std.compute() anrufen, entstehen Ihnen sowohl die Kosten für den Anruf std() als auch die Kosten für die Erstellung df. Wenn Sie df aus einem Pandas Datenrahmen erstellt haben, dann ist dies billig, aber wenn Sie df aus einem teureren Prozess erstellt haben, wie das Einlesen von CSV-Dateien, dann kann dies sehr teuer sein.

df = dd.from_pandas() # ok to call compute many times, data is already in memory 
df = dd.read_csv(...) # slow to call compute many times, we read the all the csv files every time you call compute 

Wenn Sie den Speicher haben, dann können Sie diese wiederholt Kosten vermeiden, indem

df = df.persist() 

In Ihrer Frage beharren rufen Sie sagen, dass Sie nicht bestehen bleiben können, weil Sie planen, dies zu tun multi- Arbeitersysteme. Um klar zu sein, wenn Sie über den verfügbaren Arbeitsspeicher verfügen, bleiben sie in allen Fällen sowohl für Einzelarbeiter als auch für Mehrfacharbeiter bestehen.

Sie können wiederholte Anrufe auch vermeiden, indem Sie compute einmal aufrufen.

stds = [df[column].std() for column in df.columns] 
stds = dask.compute(stds) 

Dieser berechnet alles in einem einzigen Durchgang