dask-parallel-computing

Dask — Parallel & Distributed Computing

Overview

Dask is a Python library for parallel and distributed computing that scales familiar pandas/NumPy APIs to larger-than-memory datasets. It provides five main components (DataFrames, Arrays, Bags, Futures, Schedulers) and scales from single-machine multi-core to multi-node HPC clusters.

When to Use

• Processing datasets that exceed available RAM (10 GB–100 TB)
• Parallelizing pandas or NumPy operations across multiple cores
• Processing multiple files efficiently (CSV, Parquet, JSON, HDF5, Zarr)
• Building custom parallel workflows with task dependencies
• Distributing workloads across HPC clusters (SLURM, Kubernetes)
• Streaming/ETL pipelines for unstructured data (logs, JSON records)
• For in-memory single-machine speed: use polars instead
• For out-of-core single-machine analytics: use vaex instead

Prerequisites

pip install dask[complete]        # All components
pip install dask[dataframe]       # DataFrames only
pip install dask[distributed]     # Distributed scheduler + dashboard
pip install dask-jobqueue          # HPC cluster integration (SLURM, PBS)

Core API

1. DataFrames — Parallel Pandas

import dask.dataframe as dd

# Read multiple files as a single DataFrame
ddf = dd.read_csv('data/2024-*.csv')
ddf = dd.read_parquet('data/', columns=['id', 'value', 'category'])

# Operations are lazy until .compute()
filtered = ddf[ddf['value'] > 100]
result = filtered.groupby('category').agg({'value': ['mean', 'sum']}).compute()
print(result.shape)  # (n_categories, 2)

# Custom operations via map_partitions (preferred over apply)
def normalize_partition(df):
    df['norm_value'] = (df['value'] - df['value'].mean()) / df['value'].std()
    return df

ddf = ddf.map_partitions(normalize_partition)

# Joins
ddf_merged = ddf.merge(lookup_ddf, on='category', how='left')

# Write results
ddf.to_parquet('output/', engine='pyarrow')

# Repartitioning for optimal chunk sizes
ddf = ddf.repartition(npartitions=20)         # By count
ddf = ddf.repartition(partition_size='100MB')  # By size

# Index management for sorted operations
ddf = ddf.set_index('timestamp', sorted=True)

# Debugging
print(f"Partitions: {ddf.npartitions}")
print(f"Dtypes: {ddf.dtypes}")
sample = ddf.get_partition(0).compute()  # Inspect first partition

2. Arrays — Parallel NumPy

import dask.array as da
import numpy as np

# Create from various sources
x = da.random.random((100000, 1000), chunks=(10000, 1000))
x = da.from_array(np_array, chunks=(10000, 1000))
x = da.from_zarr('large_dataset.zarr')

# Standard operations (lazy)
y = (x - x.mean(axis=0)) / x.std(axis=0)  # Normalize
z = da.dot(x.T, x)                          # Matrix multiply
u, s, v = da.linalg.svd(x)                  # SVD

# Compute and persist
result = y.mean(axis=0).compute()
print(result.shape)  # (1000,)

# Custom operations with map_blocks
def custom_filter(block):
    from scipy.ndimage import gaussian_filter
    return gaussian_filter(block, sigma=2)

filtered = da.map_blocks(custom_filter, x, dtype=x.dtype)

# Rechunking for different access patterns
x_rechunked = x.rechunk({0: 5000, 1: 500})

# Save to disk
da.to_zarr(y, 'normalized.zarr')

3. Bags — Unstructured Data Processing

import dask.bag as db
import json

# Read unstructured data
bag = db.read_text('logs/*.json').map(json.loads)

# Functional operations
valid = bag.filter(lambda x: x['status'] == 'success')
ids = valid.pluck('user_id')
flat = bag.map(lambda x: x['tags']).flatten()

# Aggregation — use foldby instead of groupby (much faster)
counts = bag.foldby(
    key='category',
    binop=lambda total, x: total + x['amount'],
    initial=0,
    combine=lambda a, b: a + b,
    combine_initial=0
).compute()

# Convert to DataFrame for structured analysis
ddf = valid.to_dataframe(meta={'user_id': 'str', 'amount': 'float64', 'category': 'str'})

4. Futures — Task-Based Parallelism

from dask.distributed import Client

client = Client()  # Local cluster with all cores
print(client.dashboard_link)  # http://localhost:8787

# Submit individual tasks (executes immediately, not lazy)
def process(x, param):
    return x ** param

future = client.submit(process, 42, param=2)
print(future.result())  # 1764

# Map over many inputs
futures = client.map(process, range(100), param=2)
results = client.gather(futures)
print(len(results))  # 100

# Scatter large data to workers (avoids repeated transfers)
import numpy as np
big_data = np.random.random((10000, 1000))
data_future = client.scatter(big_data, broadcast=True)

# Submit tasks using scattered data
futures = [client.submit(process_chunk, data_future, i) for i in range(10)]
results = client.gather(futures)

# Progressive result processing
from dask.distributed import as_completed
for future in as_completed(futures):
    result = future.result()
    print(f"Completed: {result}")

# Coordination primitives
from dask.distributed import Lock, Queue, Event
lock = Lock('resource-lock')
with lock:
    # Thread-safe operation across workers
    pass

client.close()

5. Schedulers & Configuration

import dask

# Global scheduler setting
dask.config.set(scheduler='threads')       # Default: GIL-releasing numeric work
dask.config.set(scheduler='processes')     # Pure Python, GIL-bound work
dask.config.set(scheduler='synchronous')   # Debugging with pdb

# Context manager for temporary change
with dask.config.set(scheduler='synchronous'):
    result = computation.compute()  # Can use pdb here

# Per-compute override
result = ddf.mean().compute(scheduler='processes')

# Distributed scheduler with resource control
from dask.distributed import Client
client = Client(n_workers=4, threads_per_worker=2, memory_limit='4GB')
print(client.dashboard_link)

# HPC cluster integration
from dask_jobqueue import SLURMCluster
from dask.distributed import Client

cluster = SLURMCluster(
    cores=24, memory='100GB',
    walltime='02:00:00', queue='regular'
)
cluster.scale(jobs=10)  # Request 10 SLURM jobs
client = Client(cluster)

# Adaptive scaling
cluster.adapt(minimum=2, maximum=20)

result = computation.compute()
client.close()

Key Concepts

Component Selection Guide

| Data Type | Component | When to Use | |-----------|-----------|-------------| | Tabular (CSV, Parquet) | DataFrames | Standard pandas-like operations at scale | | Numeric arrays (HDF5, Zarr) | Arrays | NumPy operations, linear algebra, image processing | | Text, JSON, logs | Bags | ETL/cleaning → convert to DataFrame for analysis | | Custom parallel tasks | Futures | Dynamic workflows, parameter sweeps, task dependencies | | Any of above | Schedulers | Control execution backend (threads/processes/distributed) |

Control level: DataFrames/Arrays/Bags = high-level lazy API. Futures = low-level immediate execution.

Lazy Evaluation Model

All DataFrames, Arrays, and Bags build a task graph — nothing executes until .compute() or .persist().

• .compute() — execute and return result to local memory
• .persist() — execute and keep result on workers (for reuse across multiple computations)
• dask.compute(a, b, c) — compute multiple results in a single pass (shares intermediates)

Chunk Size Strategy

Target: ~100 MB per chunk (or 10 chunks per core in worker memory).

| Chunk Size | Effect | |-----------|--------| | Too large (>1 GB) | Memory overflow, poor parallelization | | Optimal (~100 MB) | Good parallelism, manageable memory | | Too small (<1 MB) | Excessive scheduling overhead |

Example: 8 cores, 32 GB RAM → target ~400 MB per chunk (32 GB / 8 cores / 10).

Scheduler Selection Guide

| Scheduler | Overhead | Best For | GIL | |-----------|----------|----------|-----| | threads (default) | ~10 µs/task | NumPy, pandas, scikit-learn | Affected | | processes | ~10 ms/task | Pure