Tracking Dask Task Metrics during a Run

Wrap each computation in a performance report, and keep a compact summary of task timings:

from dask.distributed import Client, performance_report

client = Client(cluster)
with performance_report(filename=f"reports/{run_id}.html"):
    ndvi_median = cube_ndvi.median("time").compute()

The Dask dashboard shows a run while it happens and then forgets it. Performance reports and a small scheduler plugin keep that information after the Dask cluster is gone, so a slow run can be diagnosed the next morning. This page belongs to monitoring and observability for raster pipelines in Cloud Execution & Orchestration.

What the Scheduler Knows

Three streams of scheduler information For every task the scheduler records its key prefix, start and stop time, worker and result size. For every worker it tracks memory use, spilled bytes and CPU. For every transfer between workers it records bytes and duration. Performance reports capture all three; a plugin can extract the parts you want to keep. Scheduler telemetry tasks prefix, duration worker, result size workers memory, spilled bytes CPU, occupancy transfers bytes between workers transfer time Task prefixes such as open_rasterio or median tell you which stage the time went to.

Task keys in Dask have a prefix naming the operation that created them — open_dataset, getitem, nanmedian, store-map. Grouping durations by prefix turns tens of thousands of tasks into a short table of where the time went: typically a large share in reading, some in the reduction and a little in writing. When a run slows down, that table shows immediately which stage changed.

Environment & Setup

Package Version pin Used for
dask[distributed] >=2023.1 Scheduler, performance reports, plugins
pandas >=2.0 Summaries and comparisons
bokeh >=3.1 Rendering performance reports 
pip install "dask[distributed]>=2023.1" "pandas>=2.0" "bokeh>=3.1"

Complete Working Example

import json
import time
from collections import defaultdict

import pandas as pd
from dask.distributed import SchedulerPlugin
from dask.utils import key_split


class TaskTimings(SchedulerPlugin):
    """Collect duration and result size per task prefix."""
    name = "task-timings"

    def __init__(self):
        self.rows = []

    def transition(self, key, start, finish, *args, **kwargs):
        if start == "processing" and finish == "memory":
            startstops = kwargs.get("startstops") or []
            compute = sum(s["stop"] - s["start"] for s in startstops if s["action"] == "compute")
            self.rows.append({"prefix": key_split(key), "seconds": compute,
                               "nbytes": kwargs.get("nbytes", 0), "worker": kwargs.get("worker")})


def install(client):
    client.register_plugin(TaskTimings())


def collect(client) -> pd.DataFrame:
    rows = client.run_on_scheduler(lambda dask_scheduler: dask_scheduler.plugins["task-timings"].rows)
    return pd.DataFrame(rows)


def summarise(df: pd.DataFrame) -> pd.DataFrame:
    g = df.groupby("prefix")
    out = pd.DataFrame({
        "tasks": g.size(),
        "total_s": g.seconds.sum().round(1),
        "p50_s": g.seconds.median().round(3),
        "p95_s": g.seconds.quantile(0.95).round(3),
        "gb_out": (g.nbytes.sum() / 1e9).round(2),
    })
    return out.sort_values("total_s", ascending=False)


def save_run_summary(run_id: str, client, wall_seconds: float):
    s = summarise(collect(client))
    s.to_csv(f"metrics/{run_id}_tasks.csv")
    info = client.scheduler_info()
    meta = {"run": run_id, "wall_s": round(wall_seconds, 1), "workers": len(info["workers"]),
             "threads": sum(w["nthreads"] for w in info["workers"].values())}
    json.dump(meta, open(f"metrics/{run_id}_meta.json", "w"))
    return s

The plugin runs inside the scheduler and sees every task transition, so it costs a little scheduler CPU per task. For graphs of a few hundred thousand tasks this is negligible; for millions, sample, or rely on performance reports instead. The summary is a few dozen rows, small enough to keep forever and compare across months.

Reading a Performance Report

Task stream patterns A healthy task stream is densely filled with compute blocks across all worker threads. A read-bound run shows long read tasks and white gaps where threads wait. A run with transfer problems shows red communication blocks between compute. A memory-bound run shows orange spill-to-disk activity and slowing progress. Task stream shapes to recognise healthy read-bound transfer-heavy spilling White gaps: threads idle. Red: inter-worker transfer. Orange: disk spill.

The task stream is the most informative panel in a report. Gaps between reads point at object-storage latency, which is addressed by larger chunks or more concurrent reads, as in benchmarking COG read throughput from object storage. Red transfer blocks mean the graph moves data between workers, often because chunks are misaligned with the reduction. Orange spill activity means workers are out of memory, which diagnosing Dask memory spills in raster workflows addresses.

Comparing Runs

A single run’s numbers are hard to judge; a history makes them obvious. Load the last few summaries, compute each prefix’s total time relative to its recent median, and flag prefixes that have more than doubled. That catches regressions such as a dependency upgrade that disabled GDAL’s range-request merging, or a catalogue change that returns twice as many items, on the first run they appear rather than after a month of slow nights.

import glob
import pandas as pd

hist = pd.concat([pd.read_csv(p, index_col=0).assign(run=p.split("/")[-1][:13])
                  for p in sorted(glob.glob("metrics/*_tasks.csv"))[-8:]])
pivot = hist.pivot_table(index="prefix", columns="run", values="total_s")
ratio = pivot.iloc[:, -1] / pivot.iloc[:, :-1].median(axis=1)
print(ratio[ratio > 2].sort_values(ascending=False))

Exporting to a Metrics System

For pipelines that run continuously rather than nightly, the scheduler exposes Prometheus metrics at /metrics on the dashboard port, including task counts by state and worker memory. Scraping them into Prometheus or a managed equivalent gives live dashboards and alerting without any custom plugin. For batch pipelines that start and stop a cluster per run, the scrape window is short and the plugin plus stored summaries are usually the more reliable record.

Keeping Reports Useful

Performance reports are self-contained HTML and can reach tens of megabytes for long runs, so store them in object storage with a lifecycle rule rather than in the log system, and link each report from the run manifest. Name them by run identifier so the report for any night’s run can be found without searching. For very long computations, splitting the job into stages with a report per stage keeps each file small enough to open quickly and makes it obvious which stage a problem belongs to. The combination of a compact CSV summary kept forever and a detailed report kept for a month covers both long-term trends and next-morning diagnosis without storing more than is needed.

Verification

The timings should account for the run The sum of compute seconds over all tasks, divided by the total thread-seconds available during the run, gives cluster utilisation. Values near one mean the timings account for the run; very low values mean threads were idle, usually waiting on reads or the scheduler. Utilisation = task seconds ÷ thread seconds task compute idle 74% utilisation Below ~40% the cluster is oversized or starved of input. 
t0 = time.time()
result = job.compute()
wall = time.time() - t0
s = save_run_summary(run_id, client, wall)
threads = sum(w["nthreads"] for w in client.scheduler_info()["workers"].values())
util = s.total_s.sum() / (wall * threads)
print(f"utilisation {util:.0%}")
assert 0 < util <= 1.05

Common Errors

The plugin records nothing

It was registered after the computation started, or on a different client. Register it before submitting work.

Performance report is empty

The computation ran outside the with block, for example because the result was already persisted. Wrap the compute or persist call itself.

Summaries from different runs do not line up

Task prefixes changed because the code changed. Compare runs of the same pipeline version, and record the version in the metadata.

Scheduler slows down with the plugin installed

The graph has millions of tiny tasks. Sample transitions or rechunk to fewer, larger tasks.

Frequently Asked Questions

Q: What is a Dask performance report? A standalone HTML file capturing the dashboard’s task stream, worker profiles, bandwidth and memory for the duration of a block of code. It persists after the cluster shuts down.

Q: How do I see which stage of a raster job was slowest? Group task durations by key prefix. Prefixes name the operation — reads, reductions, writes — so the grouped totals show where time went.

Q: Does a scheduler plugin slow the computation? Slightly, because it runs on every task transition. For typical raster graphs the overhead is negligible; for millions of tiny tasks consider sampling.

Q: Can I get Dask metrics into Prometheus? Yes, the dashboard serves Prometheus metrics at /metrics. For short-lived batch clusters, stored per-run summaries are often more reliable than scraping.