Parallel/distributed computing. Scale pandas/NumPy beyond memory, parallel DataFrames/Arrays, multi-file processing, task graphs, for larger-than-RAM datasets and parallel workflows.
SKILL.md
Dask
Overview
Dask is a Python library for parallel and distributed computing that enables three critical capabilities:
Larger-than-memory execution on single machines for data exceeding available RAM
Parallel processing for improved computational speed across multiple cores
Distributed computation supporting terabyte-scale datasets across multiple machines
Dask scales from laptops (processing ~100 GiB) to clusters (processing ~100 TiB) while maintaining familiar Python APIs.
When to Use This Skill
This skill should be used when:
Process datasets that exceed available RAM
Scale pandas or NumPy operations to larger datasets
Parallelize computations for performance improvements
Process multiple files efficiently (CSVs, Parquet, JSON, text logs)
Build custom parallel workflows with task dependencies
Distribute workloads across multiple cores or machines
Core Capabilities
Dask provides five main components, each suited to different use cases:
1. DataFrames - Parallel Pandas Operations
Purpose: Scale pandas operations to larger datasets through parallel processing.
When to Use:
Tabular data exceeds available RAM
Need to process multiple CSV/Parquet files together
Pandas operations are slow and need parallelization
Scaling from pandas prototype to production
Reference Documentation: For comprehensive guidance on Dask DataFrames, refer to references/dataframes.md which includes:
Reading data (single files, multiple files, glob patterns)
Common operations (filtering, groupby, joins, aggregations)
Custom operations with
map_partitions
Performance optimization tips
Common patterns (ETL, time series, multi-file processing)
Quick Example:
import dask.dataframe as dd
# Read multiple files as single DataFrame
ddf = dd.read_csv('data/2024-*.csv')
# Operations are lazy until compute()
filtered = ddf[ddf['value'] > 100]
result = filtered.groupby('category').mean().compute()
Key Points:
Operations are lazy (build task graph) until .compute() called
Use map_partitions for efficient custom operations
Convert to DataFrame early when working with structured data from other sources
2. Arrays - Parallel NumPy Operations
Purpose: Extend NumPy capabilities to datasets larger than memory using blocked algorithms.
When to Use:
Arrays exceed available RAM
NumPy operations need parallelization
Working with scientific datasets (HDF5, Zarr, NetCDF)
Need parallel linear algebra or array operations
Reference Documentation: For comprehensive guidance on Dask Arrays, refer to references/arrays.md which includes:
Creating arrays (from NumPy, random, from disk)
Chunking strategies and optimization
Common operations (arithmetic, reductions, linear algebra)
Custom operations with map_blocks
Integration with HDF5, Zarr, and XArray
Quick Example:
import dask.array as da
# Create large array with chunks
x = da.random.random((100000, 100000), chunks=(10000, 10000))
# Operations are lazy
y = x + 100
z = y.mean(axis=0)
# Compute result
result = z.compute()
Key Points:
Chunk size is critical (aim for ~100 MB per chunk)
Operations work on chunks in parallel
Rechunk data when needed for efficient operations
Use map_blocks for operations not available in Dask
3. Bags - Parallel Processing of Unstructured Data
Purpose: Process unstructured or semi-structured data (text, JSON, logs) with functional operations.
When to Use:
Processing text files, logs, or JSON records
Data cleaning and ETL before structured analysis
Working with Python objects that don't fit array/dataframe formats
Need memory-efficient streaming processing
Reference Documentation: For comprehensive guidance on Dask Bags, refer to references/bags.md which includes:
import dask
import dask.dataframe as dd
# Use threads for DataFrame (default, good for numeric)
ddf = dd.read_csv('data.csv')
result1 = ddf.mean().compute() # Uses threads
# Use processes for Python-heavy work
import dask.bag as db
bag = db.read_text('logs/*.txt')
result2 = bag.map(python_function).compute(scheduler='processes')
# Use synchronous for debugging
dask.config.set(scheduler='synchronous')
result3 = problematic_computation.compute() # Can use pdb
# Use distributed for monitoring and scaling
from dask.distributed import Client
client = Client()
result4 = computation.compute() # Uses distributed with dashboard
Key Points:
Threads: Lowest overhead (~10 µs/task), best for numeric work
Processes: Avoids GIL (~10 ms/task), best for Python work
Distributed: Monitoring dashboard (~1 ms/task), scales to clusters
Can switch schedulers per computation or globally
Best Practices
For comprehensive performance optimization guidance, memory management strategies, and common pitfalls to avoid, refer to references/best-practices.md. Key principles include:
Start with Simpler Solutions
Before using Dask, explore:
Better algorithms
Efficient file formats (Parquet instead of CSV)
Compiled code (Numba, Cython)
Data sampling
Critical Performance Rules
1. Don't Load Data Locally Then Hand to Dask
# Wrong: Loads all data in memory first
import pandas as pd
df = pd.read_csv('large.csv')
ddf = dd.from_pandas(df, npartitions=10)
# Correct: Let Dask handle loading
import dask.dataframe as dd
ddf = dd.read_csv('large.csv')
2. Avoid Repeated compute() Calls
# Wrong: Each compute is separate
for item in items:
result = dask_computation(item).compute()
# Correct: Single compute for all
computations = [dask_computation(item) for item in items]
results = dask.compute(*computations)
3. Don't Build Excessively Large Task Graphs
Increase chunk sizes if millions of tasks
Use map_partitions/map_blocks to fuse operations
Check task graph size: len(ddf.__dask_graph__())
4. Choose Appropriate Chunk Sizes
Target: ~100 MB per chunk (or 10 chunks per core in worker memory)
import dask.dataframe as dd
# Extract: Read data
ddf = dd.read_csv('raw_data/*.csv')
# Transform: Clean and process
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])
# Load: Aggregate and save
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')
Unstructured to Structured Pipeline
import dask.bag as db
import json
# Start with Bag for unstructured data
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'valid')
# Convert to DataFrame for structured analysis
ddf = bag.to_dataframe()
result = ddf.groupby('category').mean().compute()
Large-Scale Array Computation
import dask.array as da
# Load or create large array
x = da.from_zarr('large_dataset.zarr')
# Process in chunks
normalized = (x - x.mean()) / x.std()
# Save result
da.to_zarr(normalized, 'normalized.zarr')
Custom Parallel Workflow
from dask.distributed import Client
client = Client()
# Scatter large dataset once
data = client.scatter(large_dataset)
# Process in parallel with dependencies
futures = []
for param in parameters:
future = client.submit(process, data, param)
futures.append(future)
# Gather results
results = client.gather(futures)
Selecting the Right Component
Use this decision guide to choose the appropriate Dask component:
