Dask
Dask skaliert Python. Es sieht aus wie Pandas/NumPy, läuft aber auf Clustern. Die Updates von 2025 konzentrieren sich auf High Performance Shuffle und GPU-Integration.
Verwendungszweck
- Tabellarische Berechnungen jenseits des Arbeitsspeichers: Verarbeitung von 10GB-1TB großen CSV/Parquet-Datensätzen auf einem einzelnen Rechner oder einem Multi-Node-Cluster.
- Parallelisieren benutzerdefinierter Python-Workflows: Verwendung von
dask.delayedzur Parallelisierung beliebiger Python-Schleifen und Aufgabenabhängigkeitsgraphen. - Verteiltes maschinelles Lernen: Integration mit Scikit-learn, XGBoost und LightGBM über verteilte Worker-Knoten.
- Skalierung von NumPy und Pandas: Direkter Ersatz vertrauter APIs für berechnungen mit arrays und DataFrames, die nicht in den Arbeitsspeicher passen.
Schnellstart
import dask.dataframe as dd
# Mehrere Gigabyte große partitionierte CSV-Dateien lazy laden
df = dd.read_csv('daten/transaktionen_*.csv')
# Aggregierte Statistiken parallel über alle CPU-Kerne berechnen
ergebnis = df.groupby('kategorie').betrag.mean().compute()
print(ergebnis)
Kernkonzepte
#Out-of-Core-Verarbeitung mit Dask DataFrame
Laden partitionierter Parquet-Dateien und lazy Berechnung von Aggregationen:
import dask.dataframe as dd
from dask.distributed import Client
# Lokalen verteilten Cluster initialisieren
client = Client(n_workers=4, threads_per_worker=2, memory_limit='4GB')
print(f"Dask Dashboard läuft unter: {client.dashboard_link}")
# Multi-File-Datensatz lazy laden
df = dd.read_parquet(
's3://analytics-bucket/transaktionen/*.parquet',
columns=['kunden_id', 'betrag', 'status', 'land'],
engine='pyarrow'
)
# Berechnungsgraphen lazy erstellen
gefiltert = df[df['status'] == 'abgeschlossen']
kennzahlen = gefiltert.groupby('land')['betrag'].agg(['mean', 'sum', 'count'])
# Graphen ausführen und konkretes Pandas DataFrame zurückgeben
ergebnis_df = kennzahlen.compute()
print(ergebnis_df)
#Benutzerdefinierte Aufgabengraphen mit dask.delayed
Parallelisieren unabhängiger Funktionsausführungen:
from dask import delayed, compute
import time
@delayed
def fetch_api_daten(endpoint: str) -> dict:
time.sleep(0.5) # Simuliert I/O
return {'endpoint': endpoint, 'status': 200, 'items': [1, 2, 3]}
@delayed
def transformiere_nutzdaten(daten: dict) -> int:
return sum(daten['items'])
@delayed
def aggregiere_summen(summen: list[int]) -> int:
return sum(summen)
# Abhängigkeits-DAG erstellen
endpunkte = ['nutzer', 'bestellungen', 'inventar', 'zahlungen']
geholt = [fetch_api_daten(ep) for ep in endpunkte]
transformiert = [transformiere_nutzdaten(f) for f in geholt]
end_summe = aggregiere_summen(transformiert)
# Alle Aufgaben parallel im Cluster ausführen
gesamtergebnis = end_summe.compute()
print("Gesamtsumme:", gesamtergebnis)
#Dask Array für verteilte lineare Algebra
Manipulieren massiver mehrdimensionaler Arrays:
import dask.array as da
# 50.000 x 50.000 Array erstellen, aufgeteilt in 5.000 x 5.000 Chunks
x = da.random.normal(10, 0.1, size=(50000, 50000), chunks=(5000, 5000))
# Berechnungen lazy durchführen
y = x + x.T
z = y[::2, ::2].mean(axis=0)
# Ergebnis berechnen
ergebnis = z.compute()
print(\