Diagnosing Dask Memory Spills in Raster Workflows

When workers spill, the fastest diagnosis is arithmetic rather than instrumentation — compute the peak a single task should need and compare it with the limit:

chunk_bytes = 1024 * 1024 * 2          # 1024² pixels × 2 bytes (uint16)
live_arrays = 5                        # e.g. red, nir, mask, intermediate, result
threads = 4                            # threads per worker

peak_per_worker = chunk_bytes * live_arrays * threads
print(f"expected peak ≈ {peak_per_worker / 1e6:.0f} MB per worker")

If that number is close to — or above — half the worker’s memory limit, the spill is explained, and the fix is in the chunking rather than in the cluster. This is the diagnostic layer under Scaling Raster Processing with Dask.

Why This Arises When Scaling Raster Work

Raster arrays are large, and the operations applied to them are cheap. That combination means a Dask raster job is almost always memory-bound rather than CPU-bound, and its failure mode is memory pressure rather than slow arithmetic.

The mechanism has four stages. Managed memory rises as tasks hold chunk arrays. Above roughly 60 percent of the limit the worker starts spilling data to disk, which is slow but survivable. Above 80 percent it pauses accepting work. Above about 95 percent the nanny kills it, and every result that worker held has to be recomputed — often causing the same pressure again on whichever worker picks up the work.

What makes it confusing is that the culprit is rarely the chunk that is being processed. It is the number of arrays alive at once, which depends on the expression rather than on the data: a masked index over three bands holds the bands, the mask, one or two intermediates and the result at the same time, so a “100 MB chunk” can be a 600 MB working set per thread.

Four thresholds between healthy and killed Below sixty percent of the worker's limit, memory is managed normally. At sixty percent the worker spills data to disk. At seventy it stops accepting new tasks briefly. At eighty it pauses. At ninety-five the nanny terminates the worker and all its held results are recomputed elsewhere. Worker memory, as a fraction of the limit 0–60% — healthy: tasks run, memory saws and drains 60% — spill to disk begins (slow, survivable) 80% — worker pauses, stops accepting tasks 95% — worker killed everything it held is recomputed elsewhere Spilling is the warning; the kill is the failure. A job that spends time above 60% is already paying for it in speed. The cascade matters: one kill can push the next worker over the same line.

Environment & Setup

Package Version Why
dask / distributed ≥2023.5 Scheduler, worker memory management, dashboard
xarray ≥2023.6 Chunked raster arrays
rioxarray ≥0.15 Chunked reads from COGs
bokeh ≥3.1 Required for the dashboard 
pip install "dask[distributed]>=2023.5" "xarray>=2023.6" "rioxarray>=0.15" "bokeh>=3.1"

Complete Working Example

This helper reports what a computation should need before it is run, and what the cluster actually did afterwards — the two numbers that turn guesswork into a diagnosis.

Symptom to cause to fix, for raster memory pressure A sawtooth that climbs, a step that never drops, high unmanaged memory and heavy inter-worker transfer each point to a different cause: chunk size, a retained persist, GDAL cache or unmanaged allocations, and a reduction fighting the chunk layout. Match the symptom to the fix symptom cause fix sawtooth peaks near the limit chunk × live arrays × threads too big halve chunks or threads step up, then flat and high a persisted collection still held del it and cancel it unmanaged memory large GDAL cache or a leak outside Dask lower GDAL_CACHEMAX per process heavy red transfer bars reduction across chunked axis rechunk with time: -1 memory fine, scheduler busy too many tiny tasks larger chunks, fewer tasks Diagnose from the profile shape first — the five rows need five different changes. 
import numpy as np
import xarray as xr
from dask.distributed import Client


def expected_peak(da: xr.DataArray, *, live_arrays: int, threads_per_worker: int) -> dict:
    """Estimate peak memory per worker for a chunked computation."""
    chunk_shape = tuple(c[0] for c in da.chunks)
    itemsize = np.dtype(da.dtype).itemsize
    chunk_bytes = int(np.prod(chunk_shape)) * itemsize

    per_thread = chunk_bytes * live_arrays
    per_worker = per_thread * threads_per_worker
    return {
        "chunk_shape": chunk_shape,
        "chunk_mb": round(chunk_bytes / 1e6, 1),
        "live_arrays": live_arrays,
        "per_thread_mb": round(per_thread / 1e6, 1),
        "per_worker_mb": round(per_worker / 1e6, 1),
        "n_chunks": int(np.prod([len(c) for c in da.chunks])),
    }


def audit_cluster(client: Client) -> list[dict]:
    """What each worker is actually holding, right now."""
    info = client.scheduler_info()["workers"]
    rows = []
    for addr, w in info.items():
        limit = w.get("memory_limit") or 0
        managed = w["metrics"].get("managed_bytes", 0)
        spilled = w["metrics"].get("spilled_bytes", {}).get("memory", 0) if isinstance(
            w["metrics"].get("spilled_bytes"), dict) else w["metrics"].get("spilled_bytes", 0)
        process = w["metrics"].get("memory", 0)
        rows.append({
            "worker": addr.rsplit("/", 1)[-1],
            "threads": w.get("nthreads"),
            "limit_gb": round(limit / 1e9, 2) if limit else None,
            "process_gb": round(process / 1e9, 2),
            "managed_gb": round(managed / 1e9, 2),
            "unmanaged_gb": round(max(0, process - managed) / 1e9, 2),
            "spilled_gb": round(spilled / 1e9, 2),
            "pct_of_limit": round(100 * process / limit, 1) if limit else None,
        })
    return sorted(rows, key=lambda r: -(r["pct_of_limit"] or 0))


if __name__ == "__main__":
    client = Client(n_workers=4, threads_per_worker=2, memory_limit="4GB")

    red = rioxarray.open_rasterio("B04.tif", masked=True, chunks={"x": 1024, "y": 1024})
    print(expected_peak(red.squeeze(), live_arrays=5, threads_per_worker=2))

    ndvi = ((nir - red) / (nir + red)).where(mask)
    result = ndvi.median("time").compute()

    for row in audit_cluster(client):
        print(row)

The unmanaged_gb column is the one to read first. Memory that Dask does not know about — GDAL’s block cache, a leaked reference, a library’s internal buffer — behaves differently from managed memory: it is never spilled, never released between tasks, and it is what turns a sawtooth into a staircase.

Variant Patterns

1. Reading the profile shape

Three shapes, three different causes A sawtooth that returns to baseline is a healthy job. A monotonic climb points to retained references or unmanaged memory. A step up that never comes down is the signature of a persisted collection that is still being held after it is no longer needed. healthy sawtooth monotonic climb step and hold drains between tasks nothing to fix never returns to baseline retained refs or unmanaged memory one jump, then flat and high a persist() nobody released Diagnose from the shape before changing anything — the three causes need three different fixes.

2. The fixes, matched to the cause

# Cause: chunks too large → peak per task exceeds the budget
red = rioxarray.open_rasterio("B04.tif", masked=True, chunks={"x": 512, "y": 512})

# Cause: too many threads sharing one worker's memory
client = Client(n_workers=8, threads_per_worker=1, memory_limit="4GB")

# Cause: a persisted collection held after use
stack = stack.persist()
...
del stack                      # release it as soon as the reuse is done
client.cancel(stack)           # and tell the scheduler

# Cause: GDAL's per-process block cache competing with Dask
import os
os.environ["GDAL_CACHEMAX"] = "256"      # per process, not per machine

The last one catches teams repeatedly, because it is memory Dask cannot see or manage. Its sizing is discussed in Caching PROJ Data and GDAL Config in Containers.

3. Restructuring the graph rather than the cluster

Some spills are structural: a reduction over an axis that is split across chunks forces data to move between workers and to be held while it moves.

# Wrong: time split across chunks, so a median over time shuffles
cube = cube.chunk({"time": 4, "y": 1024, "x": 1024})

# Right: time whole within a chunk, space split
cube = cube.chunk({"time": -1, "y": 512, "x": 512})
composite = cube.median("time")          # now each chunk reduces independently

That single change often removes both the spill and most of the runtime, and it is the same argument made in Tuning Dask Chunk Sizes for Raster Cubes.

A Diagnosis Checklist

Work down this list; the first match is usually the cause.

Is the graph enormous? len(arr.__dask_graph__()) above a few hundred thousand means the scheduler is the bottleneck and every worker is holding many small results. Larger chunks fix it.

Is unmanaged memory large? If process − managed is a substantial fraction of the limit, the memory is outside Dask: GDAL cache, a library buffer, or a reference held in user code. No amount of chunk tuning helps.

Does memory drop when the computation ends? If not, something on the client is holding results — often a notebook cell’s output, or a list built inside a loop.

Is the reduction axis chunked? A reduction across the chunked axis is the classic raster spill, and it shows as high inter-worker transfer alongside the memory growth.

Are threads multiplying the peak? A worker with eight threads runs eight tasks at once and holds eight working sets. For memory-heavy raster tasks, fewer threads and more processes is usually the better trade.

Common Errors

distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting

The classic kill. Compute the expected peak first; if it exceeds half the limit, halve the chunk size before touching anything else.

The job is slower with more workers

Spilling. Adding workers adds scheduler overhead and, if each still exceeds its budget, more spilling — not more throughput.

Memory looks fine but the dashboard shows constant red transfer

Not a memory problem: chunks are misaligned with the reduction, so data moves between workers. Rechunk rather than resize.

Frequently Asked Questions

Q: Is spilling always a problem? No. A job that spills briefly during a reduction and returns to a stable baseline is working as designed. The dangerous pattern is memory that only ever climbs, because it ends in a killed worker and the recomputation of everything that worker held.

Q: Should I add more workers or more memory per worker? Neither, usually. Peak memory per worker is set by chunk size times live arrays times threads, so halving the chunk size fixes what doubling the cluster does not — and costs nothing.

Q: Why does memory keep growing even between tasks? Something is holding references: a persisted collection, a list of intermediate results, or GDAL’s block cache sized per process. Unmanaged memory that never returns is almost always outside the Dask graph.