Deinen Data-Science-Workflow mit Dask boosten: Der umfassende Guide

Als Wes McKinney mit pandas begann, galt eine Faustregel: Damit pandas optimal läuft, sollte der Arbeitsspeicher des Rechners 5–10 Mal größer sein als das zu verarbeitende Dataset. Das ließ sich 2010 leicht einhalten, aber jetzt schreiben wir 2023.

Spätestens 2020 waren reale Datasets so groß, dass sie Alltagslaptops und -rechner problemlos zum Absturz bringen konnten. Vorausschauend wurde bereits 2015 eine Lösung veröffentlicht.

Dask ist eine Open-Source-Bibliothek der Anaconda-Macher, die die Herausforderungen skalierbarer, effizienter Verarbeitung großer Datasets löst, die nicht in den Speicher einer einzelnen Maschine passen.

Dieses Tutorial gibt dir einen umfassenden Einstieg in Dask und seine wichtigsten Features, darunter Interfaces für DataFrames, Arrays und Bags.

Dask einrichten

Wie jede andere Bibliothek kannst du Dask auf drei Arten installieren: über Conda, Pip oder aus dem Quellcode.

Da dies ein Einführungsartikel zu Dask ist, lassen wir die letzte Methode weg – sie richtet sich an Maintainer.

Wenn du Anaconda nutzt, ist Dask standardmäßig enthalten (was viel über die Popularität der Bibliothek aussagt). Möchtest du es neu installieren oder aktualisieren, nutzt du den install-Befehl:

conda install daskBereitgestellt von

Die PIP-Alternative lautet:

pip install "dask[complete]"Bereitgestellt von

Mit der Erweiterung [complete] werden auch alle benötigten Abhängigkeiten von Dask installiert, sodass du NumPy, Pandas und Tornado nicht separat einrichten musst.

Ob die Installation erfolgreich war, prüfst du über die Bibliotheksversion:

import dask

dask.__version__Bereitgestellt von

Ausgabe:

'2023.5.0'Bereitgestellt von

Der Großteil deiner Arbeit mit Dask dreht sich um drei Interfaces: Dask DataFrames, Arrays und Bags. Importieren wir sie zusammen mit numpy und pandas für den Rest des Artikels:

import dask.array as da
import dask.bag as db
import dask.dataframe as dd
import numpy as np
import pandas as pdBereitgestellt von

Grundkonzepte von Dask

Auf hoher Ebene kannst du Dask als Wrapper sehen, der die Möglichkeiten klassischer Tools wie pandas, NumPy und Spark erweitert, um größere-als-der-Speicher-Datasets zu verarbeiten.

Bei großen Objekten wie Arrays (Vektoren) oder Matrizen (DataFrames), die nicht in den Speicher passen, zerlegt Dask diese in Teilstücke, sogenannte Partitionen.

Betrachten wir zum Beispiel ein Array aus 12 Zufallszahlen in NumPy und Dask:

narr = np.random.rand(12)

narrBereitgestellt von

array([0.9261154 , 0.87774082, 0.87078873, 0.22309476, 0.24575174,
      0.04182393, 0.31476305, 0.04599283, 0.62354124, 0.97597454,
      0.23923457, 0.81201211])Bereitgestellt von

darr = da.from_array(narr, chunks=3)
darrBereitgestellt von

Das Bild oben zeigt, dass das Dask-Array vier Chunks enthält, da wir chunks auf 3 gesetzt haben. Unter der Haube ist jeder Chunk wiederum ein NumPy-Array.

Schauen wir uns nun ein deutlich größeres Beispiel an. Wir erstellen zwei 10k-×-100k-Arrays (1 Milliarde Elemente) und führen in beiden Bibliotheken elementweise Multiplikation durch – inklusive Zeitmessung:

# Create the NumPy arrays
arr1 = np.random.rand(10_000, 100_000)
arr2 = np.random.rand(10_000, 100_000)

# Create the Dask arrays
darr1 = da.from_array(arr1, chunks=(1_000, 10_000))
darr2 = da.from_array(arr2, chunks=(1_000, 10_000))Bereitgestellt von

%%time

result_np = np.multiply(arr1, arr2)Bereitgestellt von

Wall time: 3.19 sBereitgestellt von

%%time

result_dask = da.multiply(darr1, darr2)Bereitgestellt von

Wall time: 94.8 msBereitgestellt von

Der obige Code zeigt die elementweise Multiplikation zweier großer Arrays mit NumPy und Dask. Wie die Ausgabe zeigt, ist Dask für diese Berechnung etwa 34-mal schneller als NumPy. Je größer Rechenaufwand und Arraygröße werden, desto stärker fällt der Performance-Gewinn aus.

Dask nutzt denselben Ansatz des Chunkings und verteilt diese Chunks auf alle verfügbaren Kerne deines Rechners auch für andere Objekttypen.

Dask DataFrames

Um den Nutzen von Dask wirklich zu spüren, brauchen wir ein großes Dataset, idealerweise über 1 GB. Das herunterzuladen ist zum Mitmachen aber nicht optimal. Nutze stattdessen dieses Skript, das ein synthetisches Dataset mit 10 Millionen Zeilen, 10 numerischen und 10 kategorialen Features erzeugt.

Stelle sicher, dass dein Rechner mindestens 12 GB RAM hat, um das Skript auszuführen.

Sobald die Datei large_dataset.csv in deinem Workspace liegt, kannst du sie mit der Funktion read_csv aus dem Dask-DataFrame-Interface (dd) laden:

import dask.dataframe as dd

dask_df = dd.read_csv("data/large_dataset.csv")

dask_df.head()Bereitgestellt von

Obwohl die Datei groß ist, wird das Ergebnis fast augenblicklich geholt. Für noch größere Dateien kannst du den Parameter `blocksize` setzen, der bestimmt, in wie viele Byte-Blöcke die Datei aufgeteilt wird.

Ähnlich wie Dask-Arrays aus vielen kleinen NumPy-Arrays bestehen, ist Dask darauf ausgelegt, viele kleine Pandas-DataFrames entlang des Zeilenindex zu handhaben.

Wie du an read_csv siehst, bleibt ein Großteil der gewohnten Syntax und Funktionalität der pandas-API in Dask erhalten. Die folgenden Codeblöcke sollten dir aus der Arbeit mit pandas vertraut vorkommen.

Spalten auswählen und elementweise Operationen

In diesem Beispiel führen wir einfache Spaltenoperationen auf unserem Dask DataFrame dask_df aus. Wir addieren die Werte der Spalte Numeric_0 zu dem Produkt aus Numeric_9 und Numeric_3. Das Ergebnis speichern wir in result.

result = (
   dask_df["Numeric_0"] + dask_df["Numeric_9"] * dask_df["Numeric_3"]
)

result.compute().head()


0    1.301960
1    1.190679
2    1.100955
3    0.758272
4    0.926729
dtype: float64Bereitgestellt von

Wie erwähnt, führt Dask diese Operationen nicht sofort aus, sondern erstellt zunächst einen Ausführungsplan, den sogenannten Task Graph. So kann Dask Berechnungen optimieren und bei Bedarf parallelisieren. Die Funktion compute() stößt die Ausführung an, und head() zeigt die ersten Zeilen des Ergebnisses.

Bedingtes Filtern

Jetzt schauen wir, wie Dask filtern kann. Wir wählen Zeilen, in denen der Wert in der Spalte "Categorical_5" gleich "A" ist.

Das funktioniert ähnlich wie in pandas, mit einem Unterschied: Dask führt die Operation lazy aus. Es bereitet den Task Graph vor und wartet mit der Ausführung, bis wir compute() aufrufen. Mit head() sehen wir die ersten Zeilen des gefilterten DataFrames.

dask_df[dask_df["Categorical_5"] == "A"].compute().head()Bereitgestellt von

Häufige beschreibende Statistiken

Als Nächstes erzeugen wir gängige Kennzahlen mit der Funktion describe() von Dask.

Sie liefert eine Auswahl beschreibender Statistiken für unseren DataFrame, darunter Mittelwert, Standardabweichung, Minimum, Maximum usw. Wie zuvor erstellt Dask beim Aufruf von describe() den Task Graph, führt ihn aber erst mit compute() aus.

dask_df.describe().compute()Bereitgestellt von

dask_df["Categorical_3"].value_counts().compute().head()Bereitgestellt von

Categorical_3
O    386038
C    385804
A    385493
P    385490
K    385116
Name: count, dtype: int64Bereitgestellt von

Mit value_counts() zählen wir außerdem die Häufigkeit der einzelnen Werte in "Categorical_3". compute() startet die Ausführung, und head() zeigt die häufigsten Werte.

Groupby

Zum Schluss gruppieren wir mit groupby() nach den Werten in "Categorical_8". Danach wählen wir die Spalte "Numeric_7" und berechnen den Mittelwert je Gruppe.

Das entspricht der Arbeitsweise in pandas – nur eben lazy. compute() startet die Berechnung, und head() zeigt die Durchschnittswerte der Spalte "Numeric_7" für die ersten Gruppen.

dask_df.groupby("Categorical_8")["Numeric_7"].mean().compute().head()


Categorical_8
A    0.498497
B    0.499767
C    0.500622
D    0.500307
E    0.499530
Name: Numeric_7, dtype: float64
Bereitgestellt von

Sieh dir diesen Abschnitt im Dask User Guide für weitere Gemeinsamkeiten zwischen pandas und Dask an.

Lazy Evaluation

Schauen wir uns nun an, warum am Ende jeder Codepassage compute steht.

Dask wertet Code lazy aus, während Pandas im eager mode sofort Ergebnisse zurückgibt.

Vergleich aus der Küche: Lazy Evaluation ist wie Zutaten vorbereiten und Gemüse vorschneiden, aber erst kurz vor dem Anrichten alles zusammenkochen. Diese Rolle übernimmt compute.

Eager Evaluation ist, als würdest du Zutaten sofort in die Pfanne werfen, sobald sie bereit sind – so ist alles gleichzeitig servierfertig.

Lazy Evaluation ist zentral für Dasks starke Performance, weil sie Folgendes ermöglicht:

1. Weniger Rechenaufwand. Ausdrücke werden nur bei Bedarf ausgewertet (wenn compute aufgerufen wird) – überflüssige Zwischenergebnisse entfallen.
2. Optimale Ressourcennutzung. Speicher und Rechenzeit werden nicht für Zwischenergebnisse verbraten, die am Ende gar nicht benötigt werden.
3. Support für große Datasets. Daten werden on the fly oder in kleineren Chunks verarbeitet – das schont den Speicher.

Die Ergebnisse von compute werden als Pandas Series/DataFrames oder NumPy-Arrays zurückgegeben, nicht als native Dask DataFrames.

>>> type(dask_df)
dask.dataframe.core.DataFrame


>>> type(
   dask_df[["Numeric_5", "Numeric_6", "Numeric_7"]].mean().compute()
)

pandas.core.series.Series
Bereitgestellt von

Der Grund: Die meisten Datenmanipulationen liefern nur Teilmengen des ursprünglichen DataFrames und sind damit deutlich kleiner. Die Parallelisierung von Dask wird dann nicht mehr benötigt, und du kannst in pandas oder NumPy weiterarbeiten.

Dask Bags und Dask Delayed für unstrukturierte Daten

Dask Bags und Dask Delayed sind zwei Komponenten der Dask-Bibliothek, die leistungsfähige Werkzeuge für un- oder semi-strukturierte Daten bereitstellen und Lazy Evaluation ermöglichen.

Während früher vor allem tabellarische Daten üblich waren, enthalten heutige Datasets oft unstrukturierte Dateien wie Bilder, Text, Videos oder Audio. Dask Bags liefert die passende Funktionalität und API, um solche Daten parallel und skalierbar zu verarbeiten.

Ein einfaches Beispiel:

import dask.bag as db

# Create a Dask Bag from a list of strings
b = db.from_sequence(["apple", "banana", "orange", "grape", "kiwi"])

# Filter the strings that start with the letter 'a'
filtered_strings = b.filter(lambda x: x.startswith("a"))

# Map a function to convert each string to uppercase
uppercase_strings = filtered_strings.map(lambda x: x.upper())

# Compute the result as a list
result = uppercase_strings.compute()

print(result)
['APPLE']Bereitgestellt von

Hier erzeugen wir einen Dask Bag b aus einer Liste von Strings. Dann filtern wir Einträge, die mit dem Buchstaben "a" beginnen, und wandeln sie mit filter() und map() in Großbuchstaben um. Abschließend berechnen wir das Ergebnis als Liste mit compute() und geben es aus.

Stell dir vor, du führst noch komplexere Operationen auf Milliarden ähnlicher Strings aus einer Textdatei aus. Ohne Lazy Evaluation und Parallelisierung von Dask Bags würdest du schnell an Grenzen stoßen. (Mehr zu Bags in der Dask-Dokumentation.)

Dask Delayed bietet noch mehr Flexibilität und bringt Lazy Evaluation und Parallelisierung in viele weitere Szenarien. Mit Dask Delayed kannst du jede native Python-Funktion mit dem Decorator @dask.delayed in ein Lazy-Objekt verwandeln.

Ein einfaches Beispiel:

%%time

import time
import dask


@dask.delayed
def process_data(x):
   # Simulate some computation
   time.sleep(1)
   return x**2


# Generate a list of inputs
inputs = range(1000)

# Apply the delayed function to each input
results = [process_data(x) for x in inputs]

# Compute the results in parallel
computed_results = dask.compute(*results)



Wall time: 42.1 s
Bereitgestellt von

Hier definieren wir eine Funktion process_data mit @dask.delayed. Sie simuliert Rechenarbeit mit 1 Sekunde Schlaf und gibt dann das Quadrat des Eingabewerts zurück.

Ohne Parallelisierung würde diese Berechnung für 1000 Eingaben über 1000 Sekunden dauern. Mit Dask Delayed und paralleler Ausführung dauerte sie nur rund 42,1 Sekunden.

Das Beispiel zeigt die Kraft von Parallelisierung: Sie verkürzt Laufzeiten deutlich, indem sie die Arbeit effizient auf mehrere Kerne oder Worker verteilt.

Genau darum geht es bei Parallelisierung.

Fazit und weiterführende Ressourcen

Dask ist eine der zentralen Bibliotheken im Datenökosystem. Es erweitert die Funktionalität beliebter Libraries wie NumPy, Pandas und Spark und ermöglicht nahtloses Arbeiten mit Datasets, die größer sind als der verfügbare Speicher.

Mit Dask Bags und Dask Delayed bringt es Parallelisierung und Lazy Evaluation in unkonventionelle Szenarien – etwa beim Arbeiten mit unstrukturierten Daten oder nativen Python-Objekten.

Für alle Details lies dir unbedingt die Dask-Dokumentation gründlich durch.

Wenn du eine umfassende Ressource suchst, um Dask zu meistern, sieh dir unseren Kurs Parallel Programming With Dask in Python an.