Kann ich .set_index() träge (oder gleichzeitig auszuführen), auf Dask Dataframes?

Question

tl; dr:

Ist es möglich, .set_index() Methode auf mehr Dask Dataframes parallel gleichzeitig? Alternativ ist es möglich, .set_index() träge auf mehrere Dask-Datenrahmen, die folglich dazu führen würde, dass die Indizes parallel gleichzeitig gesetzt werden? Hier

ist das Szenario:

• ich mehrere Zeitreihe
• Jede Zeitreihe haben gespeichert ist mehrere .csv Dateien. Jede Datei enthält Daten, die sich auf einen bestimmten Tag beziehen. Außerdem sind die Dateien in verschiedenen Ordnern (jeder Ordner enthält Daten für einen Monat)
• Jede Zeitreihe hat unterschiedliche Abtastraten
• Alle Zeitreihen haben die gleichen Spalten. Alle haben eine Spalte, die unter anderem DateTime enthält.
• Die Daten sind zu groß, um im Speicher verarbeitet werden zu können. Deshalb benutze ich Dask.
• Ich möchte alle Zeitreihen in einem einzigen Datenrahmen zusammenführen, ausgerichtet von DateTime. Dazu muss ich zunächst alle resample() alle Zeitreihen auf eine gemeinsame Abtastrate bringen. Und dann .join() alle Zeitreihen.
• .resample() kann nur auf Index angewendet werden. Daher muss ich vor dem Resampling für jede Zeitreihe .set_index() in der DateTime-Spalte angeben.
• Wenn ich .set_index() Methode auf einer Zeitserie frage, beginnt die Berechnung sofort. Das führt dazu, dass mein Code blockiert und gewartet wird. In diesem Moment kann ich, wenn ich die Auslastung meiner Maschinenressourcen überprüfe, sehen, dass viele Kerne verwendet werden, aber die Verwendung nicht über ~ 15% hinausgeht. Das lässt mich denken, dass ich im Idealfall die Methode .set_index() gleichzeitig auf mehr als eine Zeitreihe anwenden könnte.

Nach der obigen Situation erreicht, habe ich einige nicht elegante Lösungen versucht, die Anwendung von .set_index() Verfahren auf mehreren Zeitreihen parallelisieren (zum Beispiel eines multiprocessing.Pool schaffen), die nicht erfolgreich waren. Bevor Sie weitere Details zu diesen geben, gibt es einen sauberen Weg, um die obige Situation zu lösen? Wurde das obige Szenario irgendwann bei der Implementierung von Dask berücksichtigt?

Alternativ ist es möglich, .set_index() träge? Wenn .set_index() Methode träge angewendet werden könnte, würde ich eine vollständige Berechnung Grafik mit den oben beschriebenen Schritten erstellen und am Ende würde alles parallel gleichzeitig berechnet werden (denke ich).

Answer 1

Dask.dataframe muss die Min- und Max-Werte aller Partitionen des Datenframes kennen, um die Datetime-Operationen sinnvoll parallel ablaufen zu lassen. Standardmäßig werden die Daten einmal gelesen, um gute Partitionen zu finden. Wenn die Daten nicht sortiert sind, wird eine Sortierung durchgeführt (möglicherweise sehr teuer).

In Ihrem Fall klingt es so, als ob Ihre Daten bereits sortiert sind und Sie diese möglicherweise explizit bereitstellen können.Sie sollten im letzten Beispiel der für die prompte Antwort @MRocklin Danke dd.DataFrame.set_index docstring

A common case is when we have a datetime column that we know to be 
    sorted and is cleanly divided by day. We can set this index for free 
    by specifying both that the column is pre-sorted and the particular 
    divisions along which is is separated 

    >>> import pandas as pd 
    >>> divisions = pd.date_range('2000', '2010', freq='1D') 
    >>> df2 = df.set_index('timestamp', sorted=True, divisions=divisions) # doctest: +SKIP