Dask

Was macht Dask?

Übersicht

Dask ist eine Python-Bibliothek für paralleles und verteiltes Rechnen, die drei wesentliche Funktionen bietet:

• Ausführung, die größer als der Arbeitsspeicher ist auf Einzelrechnern für Daten, die den verfügbaren RAM übersteigen
• Parallele Verarbeitung für verbesserte Rechengeschwindigkeit über mehrere Kerne hinweg
• Verteilte Berechnung für Datensätze im Terabyte-Bereich auf mehreren Rechnern

Dask skaliert von Laptops (Verarbeitung von ~100 GiB) bis zu Clustern (Verarbeitung von ~100 TiB) und behält dabei vertraute Python-APIs bei.

Aktuelle Upstream-Version: dask 2026.3.0 (PyPI, März 2026). Dokumentation: docs.dask.org. Seit 2025.1.0 ist die ausdrucksbasierte DataFrame-API mit Abfrageplanung die einzige Implementierung – installieren Sie dask-expr nicht separat und setzen Sie dataframe.query-planning: False nicht.

Schnellstart

Installation

uv pip install "dask>=2025.1"

Für einen typischen pandas/NumPy-Workflow mit dem verteilten Scheduler und dem Dashboard:

uv pip install "dask[complete]"

Entfernter Objektspeicher (S3, GCS, Azure):

uv pip install s3fs    # s3://-Pfade
uv pip install gcsfs   # gs://-Pfade

Erfordert Python 3.10+ (Python 3.9 Unterstützung wurde in 2024.12 eingestellt). Für DataFrame-E/A wird PyArrow 16+ benötigt (ab dask 2026.1.2).

Wann dieses Skill verwendet werden sollte

Dieses Skill sollte verwendet werden, wenn:

• Datensätze verarbeitet werden, die den verfügbaren RAM überschreiten
• pandas- oder NumPy-Operationen auf größere Datensätze skaliert werden sollen
• Berechnungen parallelisiert werden, um die Leistung zu verbessern
• Mehrere Dateien effizient verarbeitet werden sollen (CSVs, Parquet, JSON, Textprotokolle)
• Benutzerdefinierte parallele Workflows mit Aufgabenabhängigkeiten erstellt werden sollen
• Workloads auf mehrere Kerne oder Maschinen verteilt werden sollen

Kernfähigkeiten

Dask bietet fünf Hauptkomponenten, die jeweils für unterschiedliche Anwendungsfälle geeignet sind:

1. DataFrames – Parallele Pandas-Operationen

Zweck: pandas-Operationen durch parallele Verarbeitung auf größere Datensätze skalieren.

Verwendungszweck:

• Tabellarische Daten überschreiten den verfügbaren RAM
• Mehrere CSV/Parquet-Dateien müssen gemeinsam verarbeitet werden
• pandas-Operationen sind langsam und benötigen Parallelisierung
• Skalierung von einem pandas-Prototyp in die Produktion

Referenzdokumentation: Eine umfassende Anleitung zu Dask DataFrames finden Sie unter references/dataframes.md. Dort finden Sie:

• Lesen von Daten (einzelne Dateien, mehrere Dateien, Glob-Muster)
• Häufige Operationen (Filtern, Groupby, Joins, Aggregationen)
• Benutzerdefinierte Operationen mit map_partitions
• Tipps zur Leistungsoptimierung
• Häufige Muster (ETL, Zeitreihen, Mehrfachdateiverarbeitung)

Kurzes Beispiel:

import dask.dataframe as dd

# Mehrere Dateien als einzelnes DataFrame lesen
ddf = dd.read_csv('data/2024-*.csv')

# Operationen sind faul, bis compute() aufgerufen wird
filtered = ddf[ddf['value'] > 100]
result = filtered.groupby('category').mean().compute()

Wichtige Punkte:

• Operationen sind faul (erstellen einen Aufgabengraphen), bis .compute() aufgerufen wird
• Verwenden Sie map_partitions für effiziente benutzerdefinierte Operationen
• Konvertieren Sie frühzeitig in ein DataFrame, wenn Sie mit strukturierten Daten aus anderen Quellen arbeiten

2. Arrays – Parallele NumPy-Operationen

Zweck: NumPy-Funktionen auf Datensätze erweitern, die größer als der Speicher sind, unter Verwendung blockierter Algorithmen.

Verwendungszweck:

• Arrays überschreiten den verfügbaren RAM
• NumPy-Operationen benötigen Parallelisierung
• Arbeiten mit wissenschaftlichen Datensätzen (HDF5, Zarr, NetCDF)
• Parallele lineare Algebra oder Array-Operationen erforderlich

Referenzdokumentation: Eine umfassende Anleitung zu Dask Arrays finden Sie unter references/arrays.md. Dort finden Sie:

• Erstellen von Arrays (aus NumPy, zufällig, von der Festplatte)
• Chunking-Strategien und Optimierung
• Häufige Operationen (Arithmetik, Reduktionen, lineare Algebra)
• Benutzerdefinierte Operationen mit map_blocks
• Integration mit HDF5, Zarr und XArray

Kurzes Beispiel:

import dask.array as da

# Großes Array mit Chunks erstellen
x = da.random.random((100000, 100000), chunks=(10000, 10000))

# Operationen sind faul
y = x + 100
z = y.mean(axis=0)

# Ergebnis berechnen
result = z.compute()

Wichtige Punkte:

• Die Chunk-Größe ist entscheidend (zielen Sie auf ~100 MB pro Chunk)
• Operationen arbeiten parallel an Chunks
• Daten neu chunken, wenn dies für effiziente Operationen erforderlich ist
• Verwenden Sie map_blocks für Operationen, die in Dask nicht verfügbar sind

3. Bags – Parallele Verarbeitung unstrukturierter Daten

Zweck: Verarbeiten von unstrukturierten oder semi-strukturierten Daten (Text, JSON, Protokolle) mit funktionalen Operationen.

Verwendungszweck:

• Verarbeitung von Textdateien, Protokollen oder JSON-Datensätzen
• Datenbereinigung und ETL vor der strukturierten Analyse
• Arbeiten mit Python-Objekten, die nicht in Array-/DataFrame-Formate passen
• Speichereffiziente Streaming-Verarbeitung erforderlich

Referenzdokumentation: Eine umfassende Anleitung zu Dask Bags finden Sie unter references/bags.md. Dort finden Sie:

• Lesen von Text- und JSON-Dateien
• Funktionale Operationen (Map, Filter, Fold, Groupby)
• Konvertierung in DataFrames
• Häufige Muster (Protokollanalyse, JSON-Verarbeitung, Textverarbeitung)
• Leistungsaspekte

Kurzes Beispiel:

import dask.bag as db
import json

# JSON-Dateien lesen und parsen
bag = db.read_text('logs/*.json').map(json.loads)

# Filtern und transformieren
valid = bag.filter(lambda x: x['status'] == 'valid')
processed = valid.map(lambda x: {'id': x['id'], 'value': x['value']})

# Zur Analyse in DataFrame konvertieren
ddf = processed.to_dataframe()

Wichtige Punkte:

• Für die erste Datenbereinigung verwenden, dann in DataFrame/Array konvertieren
• Verwenden Sie foldby anstelle von groupby für bessere Leistung
• Operationen sind streaming-fähig und speichereffizient
• Für komplexe Operationen in strukturierte Formate (DataFrame) konvertieren

4. Futures – Aufgabenbasierte Parallelisierung

Zweck: Erstellen von benutzerdefinierten parallelen Workflows mit feingranularer Kontrolle über Aufgabenausführung und Abhängigkeiten.

Verwendungszweck:

• Erstellen von dynamischen, sich entwickelnden Workflows
• Sofortige Aufgabenausführung erforderlich (nicht faul)
• Berechnungen hängen von Laufzeitbedingungen ab
• Implementierung benutzerdefinierter paralleler Algorithmen
• Zustandsbehaftete Berechnungen erforderlich

Referenzdokumentation: Eine umfassende Anleitung zu Dask Futures finden Sie unter references/futures.md. Dort finden Sie:

• Einrichten des verteilten Clients
• Einreichen von Aufgaben und Arbeiten mit Futures
• Aufgabenabhängigkeiten und Datenbewegung
• Erweiterte Koordination (Warteschlangen, Sperren, Ereignisse, Akteure)
• Häufige Muster (Parametersweeps, dynamische Aufgaben, iterative Algorithmen)

Kurzes Beispiel:

from dask.distributed import Client

client = Client()  # Lokalen Cluster erstellen

# Aufgaben einreichen (wird sofort ausgeführt)
def process(x):
    return x ** 2

futures = client.map(process, range(100))

# Ergebnisse sammeln
results = client.gather(futures)

client.close()

Wichtige Punkte:

• Erfordert einen verteilten Client (auch für einen einzelnen Rechner)
• Aufgaben werden sofort nach dem Einreichen ausgeführt
• Große Daten vorab streuen, um wiederholte Übertragungen zu vermeiden
• ~1 ms Overhead pro Aufgabe (nicht geeignet für Millionen winziger Aufgaben)
• Akteure für zustandsbehaftete Workflows verwenden

5. Scheduler – Ausführungs-Backends

Zweck: Steuern, wie und wo Dask-Aufgaben ausgeführt werden (Threads, Prozesse, verteilt).

Auswahl des Schedulers:

• Threads (Standard): NumPy/Pandas-Operationen, GIL-freigebende Bibliotheken, gemeinsam genutzter Speicher
• Prozesse: Reiner Python-Code, Textverarbeitung, GIL-gebundene Operationen
• Synchron: Debuggen mit pdb, Profiling, Fehlerverständnis
• Verteilt: Dashboard, Multi-Maschinen-Cluster, erweiterte Funktionen

Referenzdokumentation: Eine umfassende Anleitung zu Dask Schedulern finden Sie unter references/schedulers.md. Dort finden Sie:

• Detaillierte Scheduler-Beschreibungen und -Eigenschaften
• Konfigurationsmethoden (global, Kontextmanager, pro Berechnung)
• Leistungsaspekte und Overhead
• Häufige Muster und Fehlerbehebung
• Thread-Konfiguration für optimale Leistung

Kurzes Beispiel:

import dask
import dask.dataframe as dd

# Threads für DataFrame verwenden (Standard, gut für numerische Daten)
ddf = dd.read_csv('data.csv')
result1 = ddf.mean().compute()  # Verwendet Threads

# Prozesse für rechenintensive Python-Arbeit verwenden
import dask.bag as db
bag = db.read_text('logs/*.txt')
result2 = bag.map(python_function).compute(scheduler='processes')

# Synchron zum Debuggen verwenden
dask.config.set(scheduler='synchronous')
result3 = problematic_computation.compute()  # Kann pdb verwenden

# Verteilt für Überwachung und Skalierung verwenden
from dask.distributed import Client
client = Client()
result4 = computation.compute()  # Verwendet verteilt mit Dashboard

Wichtige Punkte:

• Threads: Geringster Overhead (~10 µs/Aufgabe), am besten für numerische Arbeit
• Prozesse: Vermeidet GIL (~10 ms/Aufgabe), am besten für Python-Arbeit
• Verteilt: Überwachungs-Dashboard (~1 ms/Aufgabe), skaliert auf Cluster
• Scheduler können pro Berechnung oder global gewechselt werden

Best Practices

Eine umfassende Anleitung zur Leistungsoptimierung, Speicherverwaltungsstrategien und häufigen Fehlern finden Sie unter references/best-practices.md. Zu den wichtigsten Grundsätzen gehören:

Mit einfacheren Lösungen beginnen

Bevor Sie Dask verwenden, erkunden Sie:

• Bessere Algorithmen
• Effiziente Dateiformate (Parquet statt CSV)
• Kompilierter Code (Numba, Cython)
• Datenstichproben

Kritische Leistungsregeln

1. Daten nicht lokal laden und dann an Dask übergeben

# Falsch: Lädt alle Daten zuerst in den Speicher
import pandas as pd
df = pd.read_csv('large.csv')
ddf = dd.from_pandas(df, npartitions=10)

# Richtig: Dask das Laden überlassen
import dask.dataframe as dd
ddf = dd.read_csv('large.csv')

2. Wiederholte compute()-Aufrufe vermeiden

# Falsch: Jeder compute() ist separat
for item in items:
    result = dask_computation(item).compute()

# Richtig: Ein einziger compute() für alle
computations = [dask_computation(item) for item in items]
results = dask.compute(*computations)

3. Keine übermäßig großen Aufgabengraphen erstellen

• Chunk-Größen erhöhen, wenn es Millionen von Aufgaben gibt
• map_partitions/map_blocks verwenden, um Operationen zu verschmelzen
• Aufgabengraphen-Größe prüfen: len(ddf.__dask_graph__())

4. Angemessene Chunk-Größen wählen

• Ziel: ~100 MB pro Chunk (oder 10 Chunks pro Kern im Arbeiterspeicher)
• Zu groß: Speicherüberlauf
• Zu klein: Planungs-Overhead

5. Das Dashboard verwenden

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Leistung überwachen, Engpässe identifizieren

Häufige Workflow-Muster

ETL-Pipeline

import dask.dataframe as dd

# Extract: Daten lesen
ddf = dd.read_csv('raw_data/*.csv')

# Transform: Bereinigen und verarbeiten
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])

# Load: Aggregieren und speichern
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')

Pipeline von unstrukturiert zu strukturiert

import dask.bag as db
import json

# Mit Bag für unstrukturierte Daten beginnen
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'valid')

# Für strukturierte Analyse in DataFrame konvertieren
ddf = bag.to_dataframe()
result = ddf.groupby('category').mean().compute()

Großflächige Array-Berechnung

import dask.array as da

# Großes Array laden oder erstellen
x = da.from_zarr('large_dataset.zarr')

# In Chunks verarbeiten
normalized = (x - x.mean()) / x.std()

# Ergebnis speichern (mode= für Überschreiben; zarr_array_kwargs für Komprimierung)
da.to_zarr(normalized, 'normalized.zarr', mode='w')

Benutzerdefinierter paralleler Workflow

from dask.distributed import Client

client = Client()

# Großen Datensatz einmal streuen
data = client.scatter(large_dataset)

# Parallel mit Abhängigkeiten verarbeiten
futures = []
for param in parameters:
    future = client.submit(process, data, param)
    futures.append(future)

# Ergebnisse sammeln
results = client.gather(futures)

Auswahl der richtigen Komponente

Verwenden Sie diesen Entscheidungsleitfaden, um die geeignete Dask-Komponente auszuwählen:

Datentyp:

• Tabellarische Daten → DataFrames
• Numerische Arrays → Arrays
• Text/JSON/Protokolle → Bags (dann in DataFrame konvertieren)
• Benutzerdefinierte Python-Objekte → Bags oder Futures

Operationstyp:

• Standard-pandas-Operationen → DataFrames
• Standard-NumPy-Operationen → Arrays
• Benutzerdefinierte parallele Aufgaben → Futures
• Textverarbeitung/ETL → Bags

Kontrollstufe:

• Hochstufig, automatisch → DataFrames/Arrays
• Niedrigstufig, manuell → Futures

Workflow-Typ:

• Statischer Berechnungsgraph → DataFrames/Arrays/Bags
• Dynamisch, sich entwickelnd → Futures

Integrationsaspekte

Dateiformate

• Effizient: Parquet, HDF5, Zarr (spaltenbasiert, komprimiert, parallelfreundlich)
• Kompatibel, aber langsamer: CSV (nur für die erste Erfassung verwenden)
• Für Arrays: HDF5, Zarr, NetCDF

Konvertierung zwischen Sammlungen

# Bag → DataFrame
ddf = bag.to_dataframe()

# DataFrame → Array (für numerische Daten)
arr = ddf.to_dask_array(lengths=True)

# Array → DataFrame
ddf = dd.from_dask_array(arr, columns=['col1', 'col2'])

Mit anderen Bibliotheken

• XArray: Umhüllt Dask-Arrays mit beschrifteten Dimensionen (Geodaten, Bildgebung)
• Dask-ML: Maschinelles Lernen mit scikit-learn-kompatiblen APIs
• Distributed: Erweitertes Cluster-Management und -Überwachung

Debugging und Entwicklung

Iterativer Entwicklungsworkflow

1. Auf kleinen Daten mit synchronem Scheduler testen:

dask.config.set(scheduler='synchronous')
result = computation.compute()  # Kann pdb verwenden, einfaches Debugging

2. Mit Threads auf einer Stichprobe validieren:

sample = ddf.head(1000)  # Kleine Stichprobe
# Logik testen, dann auf den vollständigen Datensatz skalieren

3. Mit verteiltem Scheduler für die Überwachung skalieren:

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Leistung überwachen
result = computation.compute()

Häufige Probleme

Speicherfehler:

• Chunk-Größen verringern
• persist() strategisch verwenden und nach Gebrauch löschen
• Auf Speicherlecks in benutzerdefinierten Funktionen prüfen

Langsamer Start:

• Aufgabengraph zu groß (Chunk-Größen erhöhen)
• map_partitions oder map_blocks verwenden, um Aufgaben zu reduzieren

Schlechte Parallelisierung:

• Chunks zu groß (Anzahl der Partitionen erhöhen)
• Threads mit Python-Code verwenden (auf Prozesse umstellen)
• Datenabhängigkeiten verhindern Parallelität