dask

Dask Agent Skill

Agent-optimized skill for Dask parallel computing framework.

Quick Reference

Core Collections

import dask.dataframe as dd
import dask.array as da
import dask.bag as db
from dask import delayed

# DataFrame - parallel pandas
df = dd.read_parquet('data/*.parquet')
result = df.groupby('key').value.mean().compute()

# Array - parallel NumPy
x = da.from_zarr('data.zarr')
y = (x + x.T).mean(axis=0).compute()

# Bag - parallel lists
b = db.read_text('logs/*.txt')
counts = b.map(str.split).flatten().frequencies().compute()

# Delayed - custom parallelism
@delayed
def process(x):
    return x * 2

results = [process(i) for i in range(10)]
total = delayed(sum)(results).compute()

Distributed Scheduler

from dask.distributed import Client, LocalCluster

# Local cluster (automatic)
client = Client()

# Local cluster (manual configuration)
cluster = LocalCluster(n_workers=4, threads_per_worker=2, memory_limit='4GB')
client = Client(cluster)

# Remote cluster
client = Client('scheduler-address:8786')

# Persist data in distributed memory
df = df.persist()

# Monitor progress
client.dashboard_link  # http://localhost:8787/status

GPU Acceleration

import dask_cudf
import cupy as cp

# GPU DataFrame (cuDF)
df = dask_cudf.read_parquet('data/*.parquet')
result = df.groupby('key').value.mean().compute()

# GPU Array (CuPy)
x = da.from_array(cp.random.random((10000, 10000), dtype='float32'), chunks=(1000, 1000))
y = x @ x.T
result = y.compute()

# GPU cluster
from dask_cuda import LocalCUDACluster
cluster = LocalCUDACluster()
client = Client(cluster)

HPC Deployment

from dask_jobqueue import SLURMCluster

# SLURM cluster
cluster = SLURMCluster(
    cores=24,
    processes=4,
    memory="100GB",
    walltime="02:00:00",
    queue="gpu",
    job_extra_directives=["--gres=gpu:2"]
)

cluster.scale(jobs=10)  # Submit 10 jobs
client = Client(cluster)

Common Patterns

DataFrame Operations