dask

Dask

Dask provides distributed computing in Python, mirroring familiar NumPy, Pandas, and Scikit-learn APIs with dynamic task scheduling, high-performance shuffles, and GPU acceleration.

When to Use

• Larger-than-Memory Tabular Computation: Processing 10GB-1TB CSV/Parquet datasets on a single machine or multi-node cluster.
• Parallelizing Custom Python Workflows: Using dask.delayed to parallelize arbitrary Python loops and task graphs.
• Distributed Machine Learning: Integrating with Scikit-learn, XGBoost, and LightGBM across distributed worker nodes.
• NumPy & Pandas Scaling: Drop-in familiar APIs for out-of-core array and DataFrame calculations.

Quick Start

import dask.dataframe as dd

# Read multi-gigabyte partitioned CSV files lazily
df = dd.read_csv('data/transactions_*.csv')

# Compute aggregated statistics in parallel across all CPU cores
result = df.groupby('category').amount.mean().compute()
print(result)

Core Concepts

Out-of-Core Processing with Dask DataFrame

Loading partitioned Parquet files and computing aggregations lazily:

import dask.dataframe as dd
from dask.distributed import Client

# Initialize local distributed cluster
client = Client(n_workers=4, threads_per_worker=2, memory_limit='4GB')
print(f"Dask Dashboard running at: {client.dashboard_link}")

# Lazy load multi-file dataset
df = dd.read_parquet(
    's3://analytics-bucket/transactions/*.parquet',
    columns=['customer_id', 'amount', 'status', 'country'],
    engine='pyarrow'
)

# Build computational graph lazily
filtered = df[df['status'] == 'completed']
metrics = filtered.groupby('country')['amount'].agg(['mean', 'sum', 'count'])

# Execute graph and return concrete Pandas DataFrame
result_df = metrics.compute()
print(result_df)

Custom Task Graphs with dask.delayed

Parallelizing independent function executions:

from dask import delayed, compute
import time

@delayed
def fetch_api_data(endpoint: str) -> dict:
    time.sleep(0.5) # Simulating I/O
    return {'endpoint': endpoint, 'status': 200, 'items': [1, 2, 3]}

@delayed
def transform_payload(data: dict) -> int:
    return sum(data['items'])

@delayed
def aggregate_totals(totals: list[int]) -> int:
    return sum(totals)

# Construct dependency DAG
endpoints = ['users', 'orders', 'inventory', 'payments']
fetched = [fetch_api_data(ep) for ep in endpoints]
transformed = [transform_payload(f) for f in fetched]
final_total = aggregate_totals(transformed)

# Execute all tasks in parallel across the cluster
total_result = final_total.compute()
print("Grand Total:", total_result)