Benchmarking Orchestrator Overhead for Small Geotasks

Every task run costs something before any pixels move: a state write, a result record, a heartbeat, sometimes a serialisation round trip. That cost is invisible against a ninety-second warp and dominant against a two-hundred-millisecond one, and the crossing point is what decides whether a tile should be a task. Measuring it takes twenty minutes and settles an argument that otherwise gets decided by intuition — usually wrongly, because the overhead is larger than people expect and the small tasks are the ones that feel cheap.

When to Use This Pattern#

• Per-unit work is under a few seconds — small vector tiles, metadata reads, header inspections.
• The fan-out is wide, so a per-task cost is multiplied by thousands.
• A run takes longer than the work in it appears to justify, and nobody can say where the time goes.
• A granularity decision is being made and the argument has become a matter of opinion.

Complete Working Example#

The measurement is a no-op task at several widths, against the same work done in a plain loop.

from __future__ import annotations

import time
from statistics import mean

from prefect import flow, task, get_run_logger


@task
def noop(i: int) -> int:
    return i


@task
def work(payload: bytes, ms: int) -> int:
    """Stands in for a small geotask: a fixed amount of CPU, no I/O."""
    end = time.perf_counter() + ms / 1000
    while time.perf_counter() < end:
        pass
    return len(payload)


@flow(name="overhead-probe")
def probe(width: int, work_ms: int, payload_bytes: int = 0) -> dict:
    payload = b"x" * payload_bytes
    t0 = time.perf_counter()
    noop.map(list(range(width)))
    t1 = time.perf_counter()
    work.map([payload] * width, ms=work_ms)
    t2 = time.perf_counter()

    per_task_overhead = (t1 - t0) / width
    per_task_total = (t2 - t1) / width
    result = {
        "width": width,
        "overhead_ms": per_task_overhead * 1000,
        "total_ms": per_task_total * 1000,
        "efficiency": (work_ms / 1000) / per_task_total,
    }
    get_run_logger().info("%s", result)
    return result


if __name__ == "__main__":
    for width in (100, 1_000, 5_000):
        for work_ms in (50, 200, 1_000, 5_000):
            probe(width, work_ms)

The payload parameter matters more than it looks. Task arguments and return values are serialised and, depending on configuration, persisted — so a task that takes a tile index and returns a path costs very little, while one that takes a NumPy array and returns another costs the size of both, twice, on every run:

# Cheap: identifiers in, identifiers out. The data never crosses the boundary.
@task
def warp_tile(tile: Tile, src_uri: str, dst_uri: str) -> str:
    warp(src_uri, dst_uri, tile)
    return dst_uri


# Expensive: the array is serialised on the way in and on the way out.
@task
def warp_array(arr: "np.ndarray", tile: Tile) -> "np.ndarray":
    return warp_in_memory(arr, tile)

The fix when a task is too small is not a faster orchestrator; it is a bigger unit. Batching a hundred small tiles into one task run turns a hundred bookkeeping costs into one, and the work inside the batch can still be parallel — a process pool, a thread pool, or simply a loop, depending on whether the work releases the interpreter lock. What is lost is per-tile retry and per-tile visibility, which is a real cost and usually a smaller one than spending four fifths of the run on state writes.

Parameter & Option Reference#

Verification & Testing#

Turning the measurement into a regression test keeps the granularity honest as the pipeline changes.

def test_task_overhead_is_within_budget() -> None:
    r = probe(width=500, work_ms=0)
    assert r["overhead_ms"] < 60, f"per-task overhead regressed to {r['overhead_ms']:.1f} ms"


def test_arguments_are_identifiers_not_arrays() -> None:
    sig = inspect.signature(warp_tile.fn)
    for name, param in sig.parameters.items():
        assert param.annotation in (Tile, str, int, float), (
            f"{name} is {param.annotation}: large values crossing the task boundary "
            "are serialised on every call"
        )


def test_batch_size_keeps_efficiency_above_threshold() -> None:
    r = probe(width=200, work_ms=BATCH_MS)
    assert r["efficiency"] > 0.8, "batches have become too small to be worth orchestrating"

The argument-type test is blunt and it catches a genuine and recurring regression. Somebody refactors a task to take a pre-loaded array “to avoid reading the file twice”, and the pipeline gets slower — because the file read it removed cost forty milliseconds and the serialisation it added costs four hundred. The failure mode is that the change looks like an optimisation, reviews like an optimisation, and is only measurable at the run level where nobody attributes it.

It is worth being clear about what the benchmark does and does not settle. It measures the marginal cost of a task run, which is the number the granularity argument needs, and it deliberately excludes everything that scales with the run rather than with the task — the scheduler’s own loop, the database’s contention under load, the network between workers and control plane. Those show up as a widening gap between the measured per-task cost at width one hundred and at width five thousand, which is why the probe sweeps widths rather than measuring one. A per-task cost that triples between those two widths is telling you the control plane, not the task, is the constraint, and no amount of batching within a task will fix it.

The corresponding measurement on the other side is the one people skip: how long the run takes when the same work is done in a plain loop with no orchestrator at all. It is the floor, it takes five minutes to obtain, and it converts every subsequent discussion from opinion to arithmetic. A fan-out that takes four times the loop is not necessarily wrong — retries, visibility and bounded concurrency are worth paying for — but the factor should be a number somebody chose rather than a number nobody knows.

Common Pitfalls#

• Assuming overhead is negligible. It is tens of milliseconds per task run, which is nothing at ninety seconds of work and everything at a hundred.
• Passing arrays across the task boundary. Serialisation on both sides typically costs more than the I/O it was meant to avoid.
• Benchmarking with an in-process test harness. It omits the state writes that constitute most of the overhead; measure against a real control plane.
• Reducing task size to improve visibility. More rows is not more insight past a few thousand, and it costs proportionally.
• Batching non-idempotent work. A batch retry then repeats side effects for the units that already succeeded.
• Logging per unit inside a batch. Forty thousand log lines cost real money in ingestion and are read by nobody.

Frequently Asked Questions#

Related#

• Prefect vs Dagster for GIS workloads — the models these numbers apply to
• DAG design principles for spatial ETL — the granularity rule this measures
• Offloading GDAL work to a process pool — parallelism inside one task
• Choosing tile sizes for raster partitioning — choosing the unit in the first place
• Right-sizing workers for raster mosaic jobs — the other half of the throughput question

← Prefect vs Dagster for GIS Workloads