Basic Usage - dask_setup Recipe

This notebook demonstrates the fundamental patterns for using dask_setup, including different workload types, resource management, and cluster configuration.

Key Learning Points

  • CPU, I/O, and mixed workload configurations

  • Resource detection and memory management

  • Dashboard access and monitoring

  • Safe cluster shutdown patterns

  • HPC environment integration

Setup and Imports

First, let’s import the necessary libraries and set up our environment.

# Core imports
import time
import warnings

# Scientific computing imports
import numpy as np
import dask
import dask.array as da
import xarray as xr

from dask_setup import setup_dask_client

# Small formatting helpers used below
def format_bytes(bytes_val):
    return f"{bytes_val / (1024**3):.2f} GB"

def format_duration(seconds):
    return f"{seconds:.2f}s"

# Optional imports for better display
try:
    import psutil
    PSUTIL_AVAILABLE = True
    print("psutil available for system monitoring")
except ImportError:
    PSUTIL_AVAILABLE = False
    print("psutil not available - system monitoring limited")

print("\nEnvironment setup complete!")

psutil available for system monitoring

Environment setup complete!

System Information

Let’s check our system resources and environment before setting up dask clusters.

import os

print("SYSTEM INFORMATION")
print("=" * 50)

# Basic system info
if PSUTIL_AVAILABLE:
    cpu_count = psutil.cpu_count(logical=False)  # Physical cores
    cpu_count_logical = psutil.cpu_count(logical=True)  # Including hyperthreading
    memory = psutil.virtual_memory()
    
    print(f"CPU cores (physical): {cpu_count}")
    print(f"CPU cores (logical):  {cpu_count_logical}")
    print(f"Total memory: {format_bytes(memory.total)}")
    print(f"Available memory: {format_bytes(memory.available)}")
    print(f"Memory usage: {memory.percent:.1f}%")
else:
    print("System info limited (psutil not available)")

# Check for HPC environment variables
print("\n  HPC ENVIRONMENT DETECTION")
print("=" * 50)

hpc_vars = {
    'PBS_JOBID': 'PBS Job ID',
    'SLURM_JOB_ID': 'SLURM Job ID',
    'PBS_NCPUS': 'PBS CPU count',
    'SLURM_CPUS_ON_NODE': 'SLURM CPU count',
    'PBS_MEM': 'PBS Memory allocation',
    'SLURM_MEM_PER_NODE': 'SLURM Memory per node',
    'PBS_JOBFS': 'PBS fast local storage',
    'TMPDIR': 'Temporary directory'
}

hpc_detected = False
for var, description in hpc_vars.items():
    value = os.environ.get(var)
    if value:
        print(f" {description}: {value}")
        hpc_detected = True

if not hpc_detected:
    print("ℹ  No HPC environment detected (running on local system)")
    print("   This is normal for local development")

print("\n System check complete")

SYSTEM INFORMATION
==================================================
CPU cores (physical): 2
CPU cores (logical):  4
Total memory: 15.61 GB
Available memory: 14.27 GB
Memory usage: 8.6%

  HPC ENVIRONMENT DETECTION
==================================================
ℹ  No HPC environment detected (running on local system)
   This is normal for local development

 System check complete

CPU-Intensive Workload

CPU-intensive workloads benefit from process-based parallelism where each worker runs in a separate process with one thread. This avoids Python’s GIL (Global Interpreter Lock) limitations.

print(" CPU-INTENSIVE WORKLOAD DEMONSTRATION")
print("=" * 60)

# Create CPU-optimized cluster
client, cluster, temp_dir = setup_dask_client(
    workload_type="cpu",
    max_workers=4,  # Adjust based on your system; omit to use every core
    dashboard=True
)

print(f" CPU cluster created")
print(f"   Workers: {len(client.nthreads())}")
print(f"   Temp directory: {temp_dir}")
print(f"   Dashboard: {client.dashboard_link}")

# Display cluster information
worker_info = client.scheduler_info()['workers']
if worker_info:
    first_worker = next(iter(worker_info.values()))
    print(f"   Memory per worker: ~{first_worker.get('memory_limit', 0) / (1024**3):.1f} GB")
    print(f"   Threads per worker: {first_worker.get('nthreads', 1)}")

# Create sample CPU-intensive computation
print("\n Running CPU-intensive computation...")

# Create a moderately large array for computation
x = da.random.random((5000, 5000), chunks=(1000, 1000))
print(f"   Array shape: {x.shape}")
print(f"   Array chunks: {x.chunks}")
print(f"   Number of chunks: {x.npartitions}")

# Time the computation
start_time = time.time()
result = (x + x.T).mean().compute()  # CPU-intensive: transpose and add
computation_time = time.time() - start_time

print(f"   Computation result: {result:.6f}")
print(f"   Computation time: {format_duration(computation_time)}")
print(f"   Throughput: {format_bytes(x.nbytes)} in {format_duration(computation_time)} = {format_bytes(x.nbytes/computation_time)}/s")

 CPU-INTENSIVE WORKLOAD DEMONSTRATION
============================================================

[setup_dask_client] Configuration summary
temp/spill dir: /tmp/dask-2225
Workers: 4 | threads/worker: 1 | processes: True
Memory: total ~15.6 GiB | usable ~11.6 GiB | per-worker ~2.9 GiB
Compression: spill=auto | comm=False
 CPU cluster created
   Workers: 4
   Temp directory: /tmp/dask-2225
   Dashboard: http://127.0.0.1:38957/status
   Memory per worker: ~2.9 GB
   Threads per worker: 1

 Running CPU-intensive computation...
   Array shape: (5000, 5000)
   Array chunks: ((1000, 1000, 1000, 1000, 1000), (1000, 1000, 1000, 1000, 1000))
   Number of chunks: 25

   Computation result: 0.999942
   Computation time: 0.67s
   Throughput: 0.19 GB in 0.67s = 0.28 GB/s

client

Client

Client-7a36971a-bd41-11f1-88b1-6045bd34f621

Connection method: Cluster object Cluster type: distributed.LocalCluster
Dashboard: http://127.0.0.1:38957/status

Cluster Info

 
# Clean up
client.close()
cluster.close()
print("\n CPU cluster cleaned up")

 CPU cluster cleaned up

I/O-Intensive Workload

workload_type="io" runs one process with many threads (8–16). Threads only run in parallel when the code they execute releases Python’s GIL, so the rule is about the GIL rather than about whether the work “feels like” I/O:

  • Zarr, cloud object storage, NumPy/BLAS release the GIL → "io" works well

  • NetCDF / HDF5 reads serialise on a process-wide lock → use "cpu" instead, even though it is file reading

"io" always starts exactly one worker, so max_workers has no effect with it.

print(" I/O-INTENSIVE WORKLOAD DEMONSTRATION")
print("=" * 60)

# Create I/O-optimized cluster  
client, cluster, temp_dir = setup_dask_client(
    workload_type="io",  # always 1 worker process with 8-16 threads
    dashboard=True
)

print(f" I/O cluster created")
print(f"   Workers: {len(client.nthreads())}")
print(f"   Temp directory: {temp_dir}")
print(f"   Dashboard: {client.dashboard_link}")

# Display cluster information
worker_info = client.scheduler_info()['workers']
if worker_info:
    first_worker = next(iter(worker_info.values()))
    print(f"   Memory per worker: ~{first_worker.get('memory_limit', 0) / (1024**3):.1f} GB")
    print(f"   Threads per worker: {first_worker.get('nthreads', 1)}")

# Simulate I/O-intensive work by creating and processing multiple smaller arrays
print("\n Running I/O-intensive simulation...")

# Create multiple smaller arrays (simulating reading many files)
arrays = []
for i in range(20):  # Simulate 20 files
    arr = da.random.random((500, 500), chunks=(250, 250))
    arrays.append(arr)

print(f"   Created {len(arrays)} arrays (simulating file reads)")
print(f"   Each array shape: {arrays[0].shape}")

# Process all arrays (simulating file processing)
start_time = time.time()

# Compute means of all arrays (I/O-like pattern)
means = [arr.mean() for arr in arrays]
results = da.stack(means).compute()

computation_time = time.time() - start_time

print(f"   Processed {len(arrays)} arrays")
print(f"   Mean of means: {results.mean():.6f}")
print(f"   Processing time: {format_duration(computation_time)}")
print(f"   Throughput: {len(arrays)} arrays in {format_duration(computation_time)} = {len(arrays)/computation_time:.1f} arrays/s")

 I/O-INTENSIVE WORKLOAD DEMONSTRATION
============================================================

[setup_dask_client] Configuration summary
temp/spill dir: /tmp/dask-2225
Workers: 1 | threads/worker: 4 | processes: False
Memory: total ~15.6 GiB | usable ~11.6 GiB | per-worker ~11.6 GiB
Compression: spill=auto | comm=False
 I/O cluster created
   Workers: 1
   Temp directory: /tmp/dask-2225
   Dashboard: http://10.1.0.34:38497/status
   Memory per worker: ~11.6 GB
   Threads per worker: 4

 Running I/O-intensive simulation...
   Created 20 arrays (simulating file reads)
   Each array shape: (500, 500)
   Processed 20 arrays
   Mean of means: 0.499844
   Processing time: 0.17s
   Throughput: 20 arrays in 0.17s = 118.2 arrays/s

client

Client

Client-7ac46833-bd41-11f1-88b1-6045bd34f621

Connection method: Cluster object Cluster type: distributed.LocalCluster
Dashboard: http://10.1.0.34:38497/status

Cluster Info

 
# Clean up
client.close()
cluster.close()
print("\n I/O cluster cleaned up")

 I/O cluster cleaned up

Mixed Workload

Mixed workloads balance between CPU and I/O operations. This configuration provides a compromise between the two extremes.

print("  MIXED WORKLOAD DEMONSTRATION")
print("=" * 60)

# Create mixed workload cluster
client, cluster, temp_dir = setup_dask_client(
    workload_type="mixed",
    max_workers=3,
    dashboard=True
)

print(f"✅ Mixed cluster created")
print(f"   Workers: {len(client.nthreads())}")
print(f"   Temp directory: {temp_dir}")
print(f"   Dashboard: {client.dashboard_link}")

# Display cluster information
worker_info = client.scheduler_info()['workers']
if worker_info:
    first_worker = next(iter(worker_info.values()))
    print(f"   Memory per worker: ~{first_worker.get('memory_limit', 0) / (1024**3):.1f} GB")
    print(f"   Threads per worker: {first_worker.get('nthreads', 1)}")

# Demonstrate mixed computation and "I/O"
print("\n Running mixed workload...")

# Step 1: "Load" data (simulate reading multiple datasets)
datasets = []
for i in range(5):
    data = da.random.random((1000, 1000), chunks=(500, 500))
    datasets.append(data)

print(f"   'Loaded' {len(datasets)} datasets")

# Step 2: Process each dataset (CPU-intensive)
start_time = time.time()

processed_results = []
for i, dataset in enumerate(datasets):
    # Simulate some computation on each dataset
    result = (dataset * 2 + 1).mean()  # Simple computation
    processed_results.append(result)
    print(f"   Processed dataset {i+1}/{len(datasets)}")

# Step 3: Combine results
final_result = da.stack(processed_results).mean().compute()

computation_time = time.time() - start_time

print(f"   Final result: {final_result:.6f}")
print(f"   Total processing time: {format_duration(computation_time)}")

# Clean up
client.close()
cluster.close()
print("\n🧹 Mixed cluster cleaned up")

  MIXED WORKLOAD DEMONSTRATION
============================================================

[setup_dask_client] Configuration summary
temp/spill dir: /tmp/dask-2225
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
✅ Mixed cluster created
   Workers: 2
   Temp directory: /tmp/dask-2225
   Dashboard: http://127.0.0.1:38827/status
   Memory per worker: ~5.8 GB
   Threads per worker: 2

 Running mixed workload...
   'Loaded' 5 datasets
   Processed dataset 1/5
   Processed dataset 2/5
   Processed dataset 3/5
   Processed dataset 4/5
   Processed dataset 5/5

   Final result: 1.999570
   Total processing time: 0.47s

🧹 Mixed cluster cleaned up

Monitoring and Dashboard Usage

The Dask dashboard is crucial for monitoring cluster performance and debugging issues. Let’s explore how to use it effectively.

print(" DASHBOARD AND MONITORING DEMONSTRATION")
print("=" * 60)

# Create a cluster for monitoring demonstration
client, cluster, temp_dir = setup_dask_client(
    workload_type="cpu",
    max_workers=2,
    dashboard=True
)

print(f" Monitoring cluster created")
print(f" Dashboard URL: {client.dashboard_link}")
print(f"    Click the link above to open the dashboard in a new tab")

# Show scheduler info
scheduler_info = client.scheduler_info()
print(f"\n  Cluster Status:")
print(f"   Scheduler address: {scheduler_info['address']}")
print(f"   Workers: {len(client.nthreads())}")

for worker_id, worker_info in scheduler_info['workers'].items():
    print(f"   Worker {worker_id[:8]}:")
    #print(f"     Address: {worker_info['address']}")
    print(f"     Threads: {worker_info['nthreads']}")
    print(f"     Memory: {format_bytes(worker_info['memory_limit'])}")

# Run a computation while dashboard is available
print(f"\n Running computation (monitor in dashboard)...")
print(f"    In the dashboard, go to:")
print(f"      - Status: Overall cluster health")
print(f"      - Task Stream: Real-time task execution")
print(f"      - Memory: Worker memory usage")
print(f"      - Workers: Individual worker status")

# Create a longer-running computation for dashboard observation
large_array = da.random.random((8000, 8000), chunks=(1000, 1000))
print(f"   Array size: {format_bytes(large_array.nbytes)}")
print(f"   Chunks: {large_array.chunks}")

# Run computation
start_time = time.time()
result = large_array.sum().compute()
computation_time = time.time() - start_time

print(f"   Result: {result:.2e}")
print(f"   Time: {format_duration(computation_time)}")
print(f"\n    Check the dashboard to see the task execution pattern!")

# Keep cluster alive for a moment to allow dashboard inspection
print(f"\n⏱  Cluster will remain active for 10 seconds for dashboard inspection...")
time.sleep(10)

# Clean up
client.close()
cluster.close()
print("\n Monitoring cluster cleaned up")

 DASHBOARD AND MONITORING DEMONSTRATION
============================================================

[setup_dask_client] Configuration summary
temp/spill dir: /tmp/dask-2225
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
 Monitoring cluster created
 Dashboard URL: http://127.0.0.1:41317/status
    Click the link above to open the dashboard in a new tab

  Cluster Status:
   Scheduler address: tcp://127.0.0.1:36177
   Workers: 2
   Worker tcp://12:
     Threads: 1
     Memory: 5.81 GB
   Worker tcp://12:
     Threads: 1
     Memory: 5.81 GB

 Running computation (monitor in dashboard)...
    In the dashboard, go to:
      - Status: Overall cluster health
      - Task Stream: Real-time task execution
      - Memory: Worker memory usage
      - Workers: Individual worker status
   Array size: 0.48 GB
   Chunks: ((1000, 1000, 1000, 1000, 1000, 1000, 1000, 1000), (1000, 1000, 1000, 1000, 1000, 1000, 1000, 1000))

   Result: 3.20e+07
   Time: 0.71s

    Check the dashboard to see the task execution pattern!

⏱  Cluster will remain active for 10 seconds for dashboard inspection...

 Monitoring cluster cleaned up

Xarray Integration

dask_setup works excellently with xarray for scientific data analysis. Let’s see how to use them together.

print(" XARRAY INTEGRATION DEMONSTRATION")
print("=" * 60)

# Create cluster for xarray work
client, cluster, temp_dir = setup_dask_client(
    workload_type="mixed",  # Good for xarray's mixed I/O and computation
    max_workers=3,
    dashboard=True
)

print(f" Xarray cluster created")
print(f"   Dashboard: {client.dashboard_link}")

# Create synthetic climate-like dataset
print(f"\n Creating synthetic climate dataset...")

# Create coordinates
times = np.arange('2000-01-01', '2005-01-01', dtype='datetime64[D]')
lat = np.linspace(-90, 90, 180)
lon = np.linspace(-180, 180, 360)

print(f"   Time steps: {len(times)}")
print(f"   Spatial grid: {len(lat)} × {len(lon)}")

# Create dask arrays for the data
# Temperature with seasonal cycle
temp_data = da.random.random((len(times), len(lat), len(lon)), chunks=(365, 90, 180))
temp_data = temp_data * 50 - 10 + 273.15  # Realistic temperature range in Kelvin

# Add seasonal cycle
day_of_year = np.arange(len(times)) % 365
seasonal_cycle = 10 * np.cos(2 * np.pi * day_of_year / 365.25)
temp_data = temp_data + seasonal_cycle[:, None, None]

# Create xarray dataset
ds = xr.Dataset({
    'temperature': (['time', 'lat', 'lon'], temp_data, {
        'units': 'K',
        'long_name': 'Near-surface air temperature'
    })
}, coords={
    'time': times,
    'lat': ('lat', lat, {'units': 'degrees_north'}),
    'lon': ('lon', lon, {'units': 'degrees_east'})
})

print(f"   Dataset created:")
print(f"     Shape: {ds.temperature.shape}")
print(f"     Chunks: {ds.temperature.chunks}")
print(f"     Size: {format_bytes(ds.temperature.nbytes)}")

# Demonstrate typical xarray operations
print(f"\n Running xarray computations...")

# 1. Global mean temperature time series
start_time = time.time()
global_mean = ds.temperature.mean(['lat', 'lon'])
global_mean_computed = global_mean.compute()
time1 = time.time() - start_time

print(f"   1. Global mean time series: {format_duration(time1)}")
print(f"      Result shape: {global_mean_computed.shape}")
print(f"      Mean temperature: {global_mean_computed.mean().values:.2f} K")

# 2. Seasonal climatology
start_time = time.time()
seasonal_clim = ds.temperature.groupby('time.season').mean('time')
seasonal_computed = seasonal_clim.compute()
time2 = time.time() - start_time

print(f"   2. Seasonal climatology: {format_duration(time2)}")
print(f"      Seasons: {list(seasonal_computed.season.values)}")

# 3. Annual mean
start_time = time.time()
annual_mean = ds.temperature.resample(time='1YE').mean()
annual_computed = annual_mean.compute()
time3 = time.time() - start_time

print(f"   3. Annual means: {format_duration(time3)}")
print(f"      Years: {len(annual_computed.time)}")

total_time = time1 + time2 + time3
print(f"\n   Total computation time: {format_duration(total_time)}")
print(f"   Data processed: {format_bytes(ds.temperature.nbytes)}")
print(f"   Throughput: {format_bytes(ds.temperature.nbytes / total_time)}/s")

# Clean up
client.close()
cluster.close()
print("\n🧹 Xarray cluster cleaned up")

 XARRAY INTEGRATION DEMONSTRATION
============================================================

[setup_dask_client] Configuration summary
temp/spill dir: /tmp/dask-2225
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
 Xarray cluster created
   Dashboard: http://127.0.0.1:38975/status

 Creating synthetic climate dataset...
   Time steps: 1827
   Spatial grid: 180 × 360
   Dataset created:
     Shape: (1827, 180, 360)
     Chunks: ((365, 365, 365, 365, 365, 2), (90, 90), (180, 180))
     Size: 0.88 GB

 Running xarray computations...

   1. Global mean time series: 1.22s
      Result shape: (1827,)
      Mean temperature: 288.15 K

   2. Seasonal climatology: 1.36s
      Seasons: ['DJF', 'JJA', 'MAM', 'SON']

   3. Annual means: 0.98s
      Years: 5

   Total computation time: 3.56s
   Data processed: 0.88 GB
   Throughput: 0.25 GB/s

🧹 Xarray cluster cleaned up

Best Practices and Tips

Here are some key best practices when using dask_setup:

print("💡 BEST PRACTICES AND TIPS")
print("=" * 60)

print("\n1. 🎯 WORKLOAD TYPE SELECTION:")
print("   • 'cpu' when the hot code holds the GIL: NetCDF/HDF5 reads, pure Python, pandas apply")
print("   • 'io' when it releases the GIL: Zarr, cloud object storage, NumPy/BLAS")
print("   • 'mixed' for pipelines that do some of both")

print("\n2. 💾 MEMORY MANAGEMENT:")
print("   • The default reserve (20% of RAM, clamped to 4-50 GB) suits most nodes")
print("   • Monitor memory usage in dashboard")
print("   • Use appropriate chunk sizes (256-512 MB)")

print("\n3. 🔧 CHUNK SIZE OPTIMIZATION:")
print("   • Too small: High overhead, poor performance")
print("   • Too large: Memory issues, poor parallelism")
print("   • Target: 256-512 MB per chunk")

# Demonstrate chunk size impact
print("\n🧪 Chunk size demonstration:")

# Create small cluster for demo
client, cluster, temp_dir = setup_dask_client(
    workload_type="cpu",
    max_workers=2,
    dashboard=False
)

# Test different chunk sizes
array_shape = (4000, 4000)
chunk_sizes = [(100, 100), (1000, 1000), (2000, 2000)]

for chunks in chunk_sizes:
    # Create array
    arr = da.random.random(array_shape, chunks=chunks)
    
    # Estimate chunk size
    chunk_bytes = chunks[0] * chunks[1] * 8  # 8 bytes per float64
    chunk_mb = chunk_bytes / (1024**2)
    
    # Time computation
    start_time = time.time()
    result = arr.mean().compute()
    comp_time = time.time() - start_time
    
    print(f"   Chunks {chunks}: {chunk_mb:.1f} MB each, {arr.npartitions} chunks, {format_duration(comp_time)}")

print("\n4. 📊 DASHBOARD USAGE:")
print("   • Status: Overall cluster health")
print("   • Task Stream: Task execution timeline")
print("   • Memory: Worker memory usage")
print("   • Progress: Long-running operation status")

print("\n5. 🖥️  HPC CONSIDERATIONS:")
print("   • Use SSH tunnels for dashboard access")
print("   • Leverage $PBS_JOBFS for temporary storage")
print("   • Monitor job resource usage")
print("   • Plan for job time limits")

print("\n6. 🚨 COMMON PITFALLS:")
print("   • Don't forget to close clusters (client.close(), cluster.close())")
print("   • Watch out for memory leaks in long-running computations")
print("   • Be careful with very large arrays that exceed memory")
print("   • Test with small data before scaling up")

client.close()
cluster.close()

print("\n✅ Best practices demonstration complete")

💡 BEST PRACTICES AND TIPS
============================================================

1. 🎯 WORKLOAD TYPE SELECTION:
   • 'cpu' when the hot code holds the GIL: NetCDF/HDF5 reads, pure Python, pandas apply
   • 'io' when it releases the GIL: Zarr, cloud object storage, NumPy/BLAS
   • 'mixed' for pipelines that do some of both

2. 💾 MEMORY MANAGEMENT:
   • The default reserve (20% of RAM, clamped to 4-50 GB) suits most nodes
   • Monitor memory usage in dashboard
   • Use appropriate chunk sizes (256-512 MB)

3. 🔧 CHUNK SIZE OPTIMIZATION:
   • Too small: High overhead, poor performance
   • Too large: Memory issues, poor parallelism
   • Target: 256-512 MB per chunk

🧪 Chunk size demonstration:

[setup_dask_client] Configuration summary
temp/spill dir: /tmp/dask-2225
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

   Chunks (100, 100): 0.1 MB each, 1600 chunks, 6.12s
   Chunks (1000, 1000): 7.6 MB each, 16 chunks, 0.12s

   Chunks (2000, 2000): 30.5 MB each, 4 chunks, 0.09s

4. 📊 DASHBOARD USAGE:
   • Status: Overall cluster health
   • Task Stream: Task execution timeline
   • Memory: Worker memory usage
   • Progress: Long-running operation status

5. 🖥️  HPC CONSIDERATIONS:
   • Use SSH tunnels for dashboard access
   • Leverage $PBS_JOBFS for temporary storage
   • Monitor job resource usage
   • Plan for job time limits

6. 🚨 COMMON PITFALLS:
   • Don't forget to close clusters (client.close(), cluster.close())
   • Watch out for memory leaks in long-running computations
   • Be careful with very large arrays that exceed memory
   • Test with small data before scaling up

✅ Best practices demonstration complete

Troubleshooting Common Issues

Here are solutions to common problems you might encounter:

print("🔧 TROUBLESHOOTING COMMON ISSUES")
print("=" * 60)

print("\n❌ Problem: 'Task requires > memory_limit'")
print("✅ Solutions:")
print("   • Reduce chunk sizes (recommend_chunks() will suggest some)")
print("   • Use fewer workers, so each gets more memory")
print("   • Lower reserve_mem_gb (the reserve is taken away from the workers)")
print("   Example: setup_dask_client(workload_type='cpu', max_workers=4)")

print("\n❌ Problem: Dashboard not accessible on HPC")
print("✅ Solutions:")
print("   • Use the SSH tunnel command setup_dask_client() prints")
print("   • Set DASK_SETUP_LOGIN_HOST so it names your site's login node")
print("   • Check firewall settings")
print("   • Ensure dashboard=True in setup_dask_client()")

print("\n❌ Problem: Slow performance")
print("✅ Solutions:")
print("   • Check chunk sizes (aim for 256-512 MB)")
print("   • Verify workload_type matches your use case")
print("   • Monitor dashboard for bottlenecks")
print("   • Consider data locality and access patterns")

print("\n❌ Problem: Out of memory errors")
print("✅ Solutions:")
print("   • Lower memory_target / memory_spill so workers spill earlier")
print("   • Use smaller chunk sizes")
print("   • Process data in batches")
print("   • Monitor memory usage in dashboard")

print("\n❌ Problem: Workers frequently restarting")
print("✅ Solutions:")
print("   • Check system resources (memory, disk space)")
print("   • Reduce memory pressure")
print("   • Check for resource contention")
print("   • Monitor system logs")

print("\n💡 Debugging Tips:")
print("   • Set DASK_SETUP_LOG_LEVEL=DEBUG to see every decision dask_setup makes")
print("   • Start with small test cases")
print("   • Monitor the dashboard during execution")
print("   • Check cluster.logs() for worker messages")
print("   • Use client.profile() for performance analysis")

🔧 TROUBLESHOOTING COMMON ISSUES
============================================================

❌ Problem: 'Task requires > memory_limit'
✅ Solutions:
   • Reduce chunk sizes (recommend_chunks() will suggest some)
   • Use fewer workers, so each gets more memory
   • Lower reserve_mem_gb (the reserve is taken away from the workers)
   Example: setup_dask_client(workload_type='cpu', max_workers=4)

❌ Problem: Dashboard not accessible on HPC
✅ Solutions:
   • Use the SSH tunnel command setup_dask_client() prints
   • Set DASK_SETUP_LOGIN_HOST so it names your site's login node
   • Check firewall settings
   • Ensure dashboard=True in setup_dask_client()

❌ Problem: Slow performance
✅ Solutions:
   • Check chunk sizes (aim for 256-512 MB)
   • Verify workload_type matches your use case
   • Monitor dashboard for bottlenecks
   • Consider data locality and access patterns

❌ Problem: Out of memory errors
✅ Solutions:
   • Lower memory_target / memory_spill so workers spill earlier
   • Use smaller chunk sizes
   • Process data in batches
   • Monitor memory usage in dashboard

❌ Problem: Workers frequently restarting
✅ Solutions:
   • Check system resources (memory, disk space)
   • Reduce memory pressure
   • Check for resource contention
   • Monitor system logs

💡 Debugging Tips:
   • Set DASK_SETUP_LOG_LEVEL=DEBUG to see every decision dask_setup makes
   • Start with small test cases
   • Monitor the dashboard during execution
   • Check cluster.logs() for worker messages
   • Use client.profile() for performance analysis

Conclusion

You’ve now learned the fundamentals of using dask_setup! Here’s a summary of what we covered:

Key Takeaways:

  1. Workload Types: Choose cpu, io, or mixed by whether your code holds the GIL (NetCDF → cpu, Zarr → io)

  2. Resource Management: Reserve adequate memory and monitor usage

  3. Dashboard: Essential for monitoring and debugging

  4. Chunk Sizes: Target 256-512 MB for optimal performance

  5. Integration: Works seamlessly with xarray and other scientific libraries

Next Steps:

  • Explore other recipes for more advanced topics

  • Try dask_setup with your own datasets

  • Experiment with different configurations

  • Check out the dashboard features in detail

Additional Resources:

  • dask_setup documentation and examples

  • Dask documentation: https://docs.dask.org/

  • Xarray documentation: https://xarray.pydata.org/

Happy computing with dask_setup! 🚀