Configuration Management — dask_setup Recipe

This notebook demonstrates advanced configuration management in dask_setup, covering:

• Using built-in profiles (climate_analysis, interactive, development, etc.)

• Building explicit DaskSetupConfig objects with full parameter control

• Loading custom YAML profiles

• Using ConfigManager to list, inspect and manage profiles

• Configuration best practices

Note: Each demo creates and closes its own cluster so resources are fully independent between examples.

Setup and Imports

import sys
import time
from pathlib import Path

import dask
import dask.array as da
import numpy as np

from dask_setup import setup_dask_client
from dask_setup.config import DaskSetupConfig
from dask_setup.config_manager import ConfigManager

print("Imports OK")

Imports OK

1. Built-in Profiles

dask_setup ships a set of pre-tuned profiles for common scientific computing patterns. Use ConfigManager to list and inspect them.

manager = ConfigManager()
profiles = manager.list_profiles()
builtin = {name: p for name, p in profiles.items() if p.builtin}

print(f"Built-in profiles ({len(builtin)}):")
for name, p in builtin.items():
    cfg = p.config
    print(f"  {name:20s}  workload={cfg.workload_type:6s}  reserve={cfg.reserve_mem_gb} GB  "
          f"adaptive={cfg.adaptive}")

Built-in profiles (5):
  climate_analysis      workload=cpu     reserve=60.0 GB  adaptive=False
  zarr_io_heavy         workload=io      reserve=40.0 GB  adaptive=False
  development           workload=mixed   reserve=8.0 GB  adaptive=False
  production            workload=mixed   reserve=80.0 GB  adaptive=True
  interactive           workload=mixed   reserve=20.0 GB  adaptive=False

# Use a built-in profile — development is safe to run anywhere
client, cluster, tmp = setup_dask_client(profile="development")

print(f"Workers: {len(client.nthreads())}")
print(f"Temp dir: {tmp}")

# Quick benchmark
t0 = time.perf_counter()
x = da.random.random((2000, 2000), chunks=(500, 500))
result = (x + x.T).mean().compute()
elapsed = time.perf_counter() - t0

print(f"Result: {result:.6f}  wall={elapsed:.2f}s")

client.close(); cluster.close()

Dask Dashboard → http://127.0.0.1:38351/status

[setup_dask_client] Configuration summary
temp/spill dir: /tmp/dask-2426
Workers: 2 | threads/worker: 2 | processes: True
Memory: total ~15.6 GiB | usable ~7.6 GiB | per-worker ~3.8 GiB
Compression: spill=auto | comm=False
Workers: 2
Temp dir: /tmp/dask-2426

Result: 0.999762  wall=0.44s

2. Explicit DaskSetupConfig Object

For reproducible, version-controlled configurations, build a DaskSetupConfig directly. Every parameter has a sensible default, so you only need to specify what differs from the base.

reserve_mem_gb in config objects: called with no config, setup_dask_client() reserves 20% of RAM (clamped to 4–50 GB). A DaskSetupConfig instead carries a flat 50 GB default, because profiles need a concrete number to save, and that is more than a laptop has. Set it explicitly in any config you build.

config = DaskSetupConfig(
    workload_type="mixed",
    max_workers=4,
    reserve_mem_gb=4.0,  # set explicitly: the config-object default is 50 GB
    spill_compression="auto",  # lz4 if installed, else snappy, else none
    memory_target=0.70,
    memory_spill=0.80,
    dashboard=True,
    description="Custom mixed workload — 4 workers",
)

print("Config object:")
print(f"  workload_type   : {config.workload_type}")
print(f"  max_workers     : {config.max_workers}")
print(f"  reserve_mem_gb  : {config.reserve_mem_gb}")
print(f"  spill_compression: {config.spill_compression}")
print(f"  memory_target   : {config.memory_target}")

client, cluster, tmp = setup_dask_client(config=config)
print(f"\nCluster started — {len(client.nthreads())} workers, tmp={tmp}")

# Demonstration workload
data = da.random.random((1500, 1500), chunks=(300, 300))
result = (data * 2 + da.sin(data)).mean().compute()
print(f"Result: {result:.6f}")

client.close(); cluster.close()

Config object:
  workload_type   : mixed
  max_workers     : 4
  reserve_mem_gb  : 4.0
  spill_compression: auto
  memory_target   : 0.7

Dask Dashboard → http://127.0.0.1:40267/status

[setup_dask_client] Configuration summary
temp/spill dir: /tmp/dask-2426
Workers: 2 | threads/worker: 2 | processes: True
Memory: total ~15.6 GiB | usable ~11.6 GiB | per-worker ~5.8 GiB
Compression: spill=auto | comm=False

Cluster started — 2 workers, tmp=/tmp/dask-2426

Result: 1.460066

3. Inline dask.config Overrides

For fine-grained Dask scheduler/worker settings, use dask.config.set() as a context manager alongside setup_dask_client.

with dask.config.set({
    "distributed.scheduler.idle-timeout": "30s",
    "distributed.worker.daemon": False,
    "distributed.worker.memory.rebalance.measure": "managed_in_memory",
}):
    print("Active overrides:")
    print(f"  idle-timeout : {dask.config.get('distributed.scheduler.idle-timeout')}")
    print(f"  worker.daemon: {dask.config.get('distributed.worker.daemon')}")

    client, cluster, tmp = setup_dask_client(
        workload_type="mixed",
        max_workers=2,
        dashboard=False,
    )

    print(f"Workers: {len(client.nthreads())}")

    data = da.random.random((1000, 1000), chunks=(200, 200))
    result = (data * 2).mean().compute()
    print(f"Result: {result:.6f}")

    client.close(); cluster.close()

Active overrides:
  idle-timeout : 30s
  worker.daemon: False

[setup_dask_client] Configuration summary
temp/spill dir: /tmp/dask-2426
Workers: 2 | threads/worker: 2 | processes: True
Memory: total ~15.6 GiB | usable ~11.6 GiB | per-worker ~5.8 GiB
Compression: spill=auto | comm=False
Workers: 2

Result: 0.999540

4. Custom YAML Profiles

Save named configurations as YAML files for sharing across a team or project. A profile keeps its settings under a config: key; configs/cpu_profile.yaml ships with this recipe as an example.

⚠️ A YAML file with its settings at the top level (no config: key) still loads, but every setting in it is ignored and you get the defaults. Check profile.config after loading.

manager = ConfigManager()
profile = manager.load_profile_from_file(Path("../configs/cpu_profile.yaml"))

print(f"Profile     : {profile.name}")
print(f"Description : {profile.description}")
cfg = profile.config
print(f"  workload_type    : {cfg.workload_type}")
print(f"  reserve_mem_gb   : {cfg.reserve_mem_gb}")
print(f"  spill_compression: {cfg.spill_compression}")
print(f"  tags             : {cfg.tags}")

# Use it, overriding what doesn't suit this machine: explicit arguments always win.
# (This profile reserves 60 GB, sized for a Gadi node, not a laptop.)
client, cluster, tmp = setup_dask_client(
    config=cfg, max_workers=2, reserve_mem_gb=4.0, dashboard=False
)
print(f"\nStarted {len(client.nthreads())} workers from '{profile.name}'")
client.close(); cluster.close()

Profile     : cpu_intensive
Description : Heavy computation workload optimized for mathematical operations
  workload_type    : cpu
  reserve_mem_gb   : 60.0
  spill_compression: auto
  tags             : ['cpu-heavy', 'mathematical', 'scientific', 'single-threaded']

[setup_dask_client] Configuration summary
temp/spill dir: /tmp/dask-2426
Workers: 2 | threads/worker: 1 | processes: True
Memory: total ~15.6 GiB | usable ~11.6 GiB | per-worker ~5.8 GiB
Compression: spill=auto | comm=False

Started 2 workers from 'cpu_intensive'

import tempfile

from dask_setup.config import ConfigProfile

# Save a profile. Normally you would use ConfigManager(), which saves to
# ~/.dask_setup/profiles/; a temporary directory keeps this demo self-contained.
demo_manager = ConfigManager(config_dir=tempfile.mkdtemp(prefix="dask_setup_profiles_"))

demo_manager.save_profile(
    ConfigProfile(
        name="zarr_pipeline",
        config=DaskSetupConfig(
            workload_type="io",  # Zarr releases the GIL, so threads read in parallel
            reserve_mem_gb=20.0,
            description="Zarr-backed analysis: one process, many threads",
            tags=["zarr", "io"],
        ),
    )
)

saved = demo_manager.config_dir / "profiles" / "zarr_pipeline.yaml"
print(f"Profile written to: {saved}\n")
print(saved.read_text())

# Profiles are looked up user -> site -> builtin, so this is now visible by name
print("Known profiles:", sorted(demo_manager.list_profiles()))

Profile written to: /tmp/dask_setup_profiles_1_9s1iw_/profiles/zarr_pipeline.yaml

builtin: false
config:
  adaptive: false
  comm_compression: false
  dashboard: true
  dashboard_port: null
  description: 'Zarr-backed analysis: one process, many threads'
  io_access_pattern: auto
  io_compression_level: null
  io_format: null
  io_storage_location: auto
  io_target_chunk_mb:
  - 128
  - 512
  max_mem_gb: null
  max_workers: null
  memory_pause: 0.92
  memory_spill: 0.85
  memory_target: 0.75
  memory_terminate: 0.98
  min_workers: null
  name: zarr_pipeline
  reserve_mem_gb: 20.0
  silence_logs: false
  spill_compression: auto
  spill_threads: null
  suggest_chunks: false
  tags:
  - zarr
  - io
  temp_base_dir: null
  workload_type: io
created_at: '2026-10-01T02:41:05.853649'
modified_at: '2026-10-01T02:41:05.853649'
name: zarr_pipeline
version: '1.7'

Known profiles: ['climate_analysis', 'development', 'interactive', 'production', 'zarr_io_heavy', 'zarr_pipeline']

5. Profile Performance Comparison

Benchmark several profiles on the same workload to pick the best fit.

profiles_to_test = ["development", "interactive"]
times = {}

for profile_name in profiles_to_test:
    try:
        client, cluster, _ = setup_dask_client(profile=profile_name, dashboard=False)

        t0 = time.perf_counter()
        x = da.random.random((2000, 2000), chunks=(500, 500))
        _ = (x + x.T).mean().compute()
        times[profile_name] = time.perf_counter() - t0

        client.close(); cluster.close()
        print(f"{profile_name:20s}  {times[profile_name]:.2f}s")
    except Exception as e:
        print(f"{profile_name:20s}  ERROR: {e}")
        times[profile_name] = float("inf")

if times:
    fastest = min(times, key=times.get)
    print(f"\nFastest: {fastest} ({times[fastest]:.2f}s)")

[setup_dask_client] Configuration summary
temp/spill dir: /tmp/dask-2426
Workers: 2 | threads/worker: 2 | processes: True
Memory: total ~15.6 GiB | usable ~7.6 GiB | per-worker ~3.8 GiB
Compression: spill=auto | comm=False

development           0.42s
interactive           ERROR: ❌ Insufficient memory for configuration:
   - Required: 1.0 GB
   - Available: 0.0 GB
   - Shortfall: 1.0 GB

💡 Suggestions:
   1. Reduce reserve_mem_gb from 20.0 GB to at most 14.6 GB
   2. Request a larger memory allocation for your job

Fastest: development (0.42s)

Configuration Best Practices

Priority order (highest wins): explicit function arguments → profile= or config= → built-in defaults. An argument you pass always wins, even when it equals the default.

Profile selection guide:

Profile	Workload	When to use
climate_analysis	cpu	Large xarray reductions on HPC; NetCDF/HDF5 input
zarr_io_heavy	io	Zarr stores and cloud object storage (not NetCDF — its reads serialise on a lock, so use cpu)
interactive	mixed	Jupyter notebooks on a login node
development	mixed	Testing on a laptop or small VM
production	mixed	Long-running automated jobs

YAML profile tips:

• Keep profiles in version control alongside your analysis scripts

• Put settings under config: and use the description and tags fields there; YAML comments work for longer notes

• Validate with dask-setup validate <name> before deploying

• Use based_on: parent_profile to inherit and override a base profile

Memory tuning:

# Conservative — for shared systems or laptops
DaskSetupConfig(reserve_mem_gb=4.0, memory_target=0.60, memory_spill=0.70)

# Aggressive — large dedicated HPC node
DaskSetupConfig(reserve_mem_gb=8.0, memory_target=0.80, spill_compression="lz4")  # needs `pip install lz4`