Coastal & Marine Spatial Analysis Pipelines

Profiling Dask Graph Overhead on Tile Workloads

A gridding job runs for four hours on a sixteen-worker cluster. The workers are at 30% CPU, the network is quiet, and the obvious conclusion is that the cluster is too small. Adding eight more workers makes it slower. The job is scheduler-bound: it has more tasks than the scheduler can dispatch, and every extra worker adds coordination rather than throughput. This page is the measurement that distinguishes that case from the others, within the Dask versus Ray comparison.

Root cause: the scheduler costs about a millisecond per task

Dask’s central scheduler tracks every task’s state, dependencies and location. That bookkeeping costs on the order of one millisecond per task, which is negligible for a task that runs for a second and dominant for one that runs for two milliseconds.

A bathymetric workload reaches large task counts easily. A 40,000 × 40,000 grid in 512-pixel chunks is 6,241 chunks; put it through eight stages and the graph is 50,000 tasks before any overlap is added. Add map_overlap for a slope kernel and each output chunk depends on nine input chunks, so the graph grows again.

import dask.array as da
import numpy as np

for chunk in (256, 512, 2048, 4096):
    x = da.zeros((40_000, 40_000), chunks=chunk)
    y = (x + 1) * 2
    y = y.map_overlap(lambda b: b, depth=1, boundary="none")
    print(f"chunk {chunk:>5}: {len(y.dask):>8,} tasks, "
          f"{x.nbytes / x.npartitions / 2**20:>6.1f} MiB per chunk")
# chunk   256:  117,406 tasks,    0.5 MiB per chunk
# chunk   512:   29,566 tasks,    2.0 MiB per chunk
# chunk  2048:    1,918 tasks,   32.0 MiB per chunk
# chunk  4096:      502 tasks,  128.0 MiB per chunk

At 256-pixel chunks the scheduler alone needs roughly two minutes just to dispatch, before a single byte is read. At 2048 it needs two seconds. The compute is identical.

Three Regimes, and Only the Middle One Is Compute A stacked area chart of total wall clock against chunk size, from a quarter of a mebibyte to five hundred mebibytes per chunk. At the small end, a scheduler-overhead band occupies most of the height, because the task count is in the hundreds of thousands and each costs about a millisecond to dispatch. Through the middle range the scheduler band shrinks to almost nothing and a compute band dominates, giving the shortest total wall clock. At the large end a spill-and-memory band grows sharply as each task's working set exceeds the worker's share of memory and the worker begins writing to disk. The total curve is a broad U with its floor spanning roughly one to sixty-four mebibytes per chunk, annotated as the range worth targeting. chunk size (MiB, log scale) wall clock scheduler compute spill the range worth targeting: 1–64 MiB 0.25132500 Adding workers moves the compute band only. On the left-hand slope it makes the total worse, because the

Step-by-step diagnosis with production code

1. Count the tasks before you run anything

The cheapest diagnostic runs in milliseconds and needs no cluster:

import logging

logger = logging.getLogger(__name__)

SCHEDULER_COST_S = 1e-3      # roughly, per task, for the distributed scheduler


def graph_budget(collection, *, expected_task_s: float) -> dict[str, float]:
    """Predicted scheduler overhead against predicted compute, before running."""
    n = len(collection.dask)
    sched_s = n * SCHEDULER_COST_S
    compute_s = n * expected_task_s
    ratio = sched_s / max(compute_s, 1e-9)
    logger.info("%d tasks: ~%.0f s scheduling vs ~%.0f s compute (%.0f%% overhead)",
                n, sched_s, compute_s, ratio * 100)
    if ratio > 0.2:
        logger.warning(
            "scheduling is >20%% of predicted compute — increase the chunk size "
            "before adding workers, which will not help")
    return {"tasks": n, "scheduler_s": sched_s, "compute_s": compute_s, "ratio": ratio}

expected_task_s is a measurement, not a guess: run one chunk through the pipeline and time it. That single number, multiplied by the task count, predicts the job well enough to catch a bad chunk size before a cluster is provisioned.

2. Split the measured wall clock three ways 

from contextlib import contextmanager


@contextmanager
def profile_run(client, path: str = "dask-report.html"):
    """Capture a performance report and log the three-way time split."""
    from distributed import performance_report
    with performance_report(filename=path):
        yield
    tasks = client.run_on_scheduler(
        lambda dask_scheduler: {
            "n": len(dask_scheduler.tasks),
            "compute_s": sum(dask_scheduler.cumulative_worker_metrics.get(("execute", k), 0.0)
                             for k in ("thread-cpu", "thread-noncpu")),
            "transfer_s": dask_scheduler.cumulative_worker_metrics.get(
                ("get-data", "network"), 0.0),
        })
    logger.info("tasks %d · compute %.0f s · transfer %.0f s · report %s",
                tasks["n"], tasks["compute_s"], tasks["transfer_s"], path)

The three numbers separate the three fixes. High compute is a real workload and wants more workers. High transfer means chunks are being shuffled between workers, and the fix is usually a chunk alignment problem rather than a size one. High residual — wall clock minus compute minus transfer — is scheduler overhead, and only a larger chunk size touches it.

3. Fuse the cheap stages

Dask fuses adjacent elementwise operations automatically, and stops fusing at anything it cannot see through — a map_blocks with an opaque function, a rechunk, a persist. A pipeline written as eight separate map_blocks calls has eight tasks per chunk where one would do:

def fused_stage(block):
    """One task per chunk instead of six. Same arithmetic, a sixth of the graph."""
    block = apply_tide(block)
    block = apply_svp(block)
    block = mask_nodata(block)
    block = to_mllw(block)
    block = clip_range(block)
    return round_to_cm(block)


result = dem.map_blocks(fused_stage, dtype="float32")

The readability cost is real and the graph cost is a factor of six. On a scheduler-bound job that is the whole difference; on a compute-bound one it changes nothing, which is why the measurement comes first.

Six Tasks per Chunk, or One Two task graphs for the same six-stage pipeline over one chunk. In the unfused version, six separate task nodes are chained, each an opaque map_blocks call the optimiser cannot see through, giving six tasks per chunk and thirty-seven thousand tasks across a six-thousand-chunk grid. In the fused version, the same six operations are inside a single Python function, so the graph holds one task per chunk and six thousand tasks in total. A note records that the arithmetic is identical, that the readability cost is real, and that this change only matters when the three-way measurement has already shown the job to be scheduler-bound. six map_blocks calls — 6 tasks/chunk tidesvpmask mllwclipround 6,241 chunks → 37,446 tasks ≈ 37 s of pure scheduling one fused call — 1 task/chunk fused_stage(block) tide · svp · mask · mllw · clip · round same arithmetic, one task 6,241 chunks → 6,241 tasks ≈ 6 s of scheduling On a scheduler-bound job this is the whole difference. On a compute-bound one it changes nothing measurable. Which is why the three-way split comes first: it says whether the readability cost buys anything at all.

Verification and acceptance test 

def test_the_graph_is_not_scheduler_bound(pipeline, single_chunk_seconds):
    budget = graph_budget(pipeline, expected_task_s=single_chunk_seconds)
    assert budget["ratio"] < 0.2, (
        f"{budget['tasks']:,} tasks would spend {budget['ratio']:.0%} of the job "
        "in the scheduler — raise the chunk size")


def test_chunks_are_in_the_useful_size_band(arr):
    per_chunk_mib = arr.nbytes / arr.npartitions / 2**20
    assert 1.0 <= per_chunk_mib <= 64.0, (
        f"{per_chunk_mib:.1f} MiB per chunk is outside the 1–64 MiB band")

Both run without a cluster, in milliseconds, and both catch the problem at the point where it is a one-line fix rather than after a four-hour run.

Reading the dashboard without guessing

The performance report and the live dashboard show the same three regimes, and each has a signature that is unambiguous once you know what to look for.

Scheduler-bound shows a task-stream panel that is mostly white space: short coloured bars separated by gaps, with every worker idle between them. The scheduler cannot feed them fast enough. The progress bar advances smoothly and slowly, and CPU utilisation sits low and flat across all workers simultaneously — the tell that distinguishes it from a slow worker, which would show one low line among several high ones.

Transfer-bound shows long red transfer bars in the task stream and a communication panel with sustained throughput. Workers are busy, but busy moving data. On a bathymetric job this almost always means chunks are misaligned between two arrays being combined, so every output chunk needs input from several workers — the alignment problem described in the tidal datum topic, seen from the scheduler’s side.

Memory-bound shows the worker memory bars crossing into orange and grey, and disk-write appearing in the task stream. Once spilling starts, throughput collapses non-linearly, so a job that is 10% over the memory budget does not run 10% slower.

def classify_bottleneck(wall_s: float, compute_s: float, transfer_s: float,
                        spill_s: float, n_workers: int) -> str:
    """Name the bottleneck from the three-way split, rather than from a feeling."""
    capacity_s = wall_s * n_workers
    unexplained = capacity_s - compute_s - transfer_s - spill_s
    fractions = {
        "compute": compute_s / capacity_s,
        "transfer": transfer_s / capacity_s,
        "spill": spill_s / capacity_s,
        "scheduler/idle": max(unexplained, 0.0) / capacity_s,
    }
    worst = max(fractions, key=fractions.get)
    logger.info("bottleneck: %s — %s", worst,
                ", ".join(f"{k} {v:.0%}" for k, v in fractions.items()))
    return worst

Dividing by wall_s * n_workers rather than by wall_s is what makes the fractions add to one. Idle worker capacity is the quantity that matters, and a job whose workers are 70% idle is not a job that needs more workers.

Three Task Streams, Three Unambiguous Signatures Three task-stream panels, each showing four worker rows over the same time window. The scheduler-bound panel shows short compute bars separated by wide gaps on every row simultaneously, so all four workers are idle at the same moments; a note says this simultaneity is what distinguishes it from one slow worker. The transfer-bound panel shows compute bars separated by long communication bars, with the workers busy but moving data rather than computing. The memory-bound panel shows compute bars becoming progressively shorter and disk-write bars appearing between them, with the note that throughput collapses non-linearly once spilling begins. scheduler-bound gaps line up across every worker → raise the chunk size transfer-bound pale bars are data movement → align the chunks memory-bound dark bars are disk spill → shrink the chunk size Two of the three fixes move the chunk size, in opposite directions. Which is why guessing does not work, and why the three-way split is worth the one function call it costs. Record the split in the run manifest: a regression between releases is otherwise invisible until someone notices the nightly job finishing later than it used to.

Edge cases and gotchas

  • len(collection.dask) counts the unoptimised graph. Dask fuses at compute time, so the real task count can be lower. dask.optimize(collection)[0].dask gives the post-fusion count and is the honest number for this estimate.
  • A large chunk that spills is worse than a small one that does not. The right-hand slope of the curve above is steep. Size chunks against the worker’s memory share, not the node’s — the same trap described in handling OOM kills.
  • Overlap multiplies the graph, not just the data. map_overlap with depth 1 adds a rechunk-like layer with several tasks per chunk. Where the overlap is the dominant cost, computing the derivative in a single fused pass over a larger chunk is cheaper than a small-chunk overlap.

Up: Dask vs Ray for Bathymetric Processing at Scale