Geospatial Orchestration Hub

Fanning Out Prefect Tasks Over a Tile Manifest

.map() turns one task into as many runs as there are items, and the three things that make that safe are all opt-in: a named concurrency limit so the upstream service is not hit by every run at once, return_state=True so one failure does not cancel the rest, and a per-item idempotency key so a re-run skips what already succeeded. Write the manifest to storage before mapping, and the fan-out becomes something you can inspect, resume and explain rather than a number that appears in the UI.

When to Use This Pattern

  • One input produces many independent outputs — a scene to tiles, a country to municipalities, a mosaic to windows.
  • The count is only known at runtime, from the source’s footprint rather than from configuration.
  • Items are individually retryable, so a failure is a tile rather than a batch.
  • Prefect is the orchestrator. The Dagster equivalent uses DynamicOut and is covered in dynamic task mapping for tile fan-out.

Complete Working Example

The manifest is written first, the limit is named, and the states are collected rather than the results.

from __future__ import annotations

import json
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Optional

from prefect import flow, task, get_run_logger
from prefect.concurrency.sync import concurrency
from prefect.states import State


@dataclass(frozen=True)
class TileItem:
    z: int
    x: int
    y: int
    source_uri: str
    source_digest: str

    @property
    def work_key(self) -> str:
        return f"{self.source_digest}:{self.z}/{self.x}/{self.y}"


@task(persist_result=True)
def write_manifest(source_uri: str, digest: str, zoom: int, path: Path) -> list[TileItem]:
    """Enumerate against the FOOTPRINT, not the bounding box, then persist."""
    items = [
        TileItem(zoom, t.x, t.y, source_uri, digest)
        for t in tiles_intersecting(footprint_of(source_uri), zoom)
    ]
    path.write_text(json.dumps([asdict(i) for i in items]))
    get_run_logger().info("manifest %s: %d tiles at z%d", path.name, len(items), zoom)
    if len(items) > 20_000:
        raise ValueError(f"fan-out of {len(items)} is implausible — check the zoom")
    return items


@task(retries=2, retry_delay_seconds=[15, 60], timeout_seconds=900, tags=["tile"])
def render_tile(item: TileItem, out_dir: Path) -> str:
    """One tile. Idempotent, bounded, and it returns a path."""
    dest = out_dir / f"{item.z}_{item.x}_{item.y}.tif"
    if ledger_has(item.work_key):
        return str(dest)                       # already done: microseconds, not minutes

    # The named limit is registered once with the Prefect server and shared by
    # every flow that reads this source — including runs we did not start.
    with concurrency("scene-reads", occupy=1):
        warp_window(item.source_uri, dest, item.z, item.x, item.y)

    ledger_mark(item.work_key)
    return str(dest)


@flow(name="tile-fanout")
def fan_out(source_uri: str, zoom: int, scratch: Path) -> dict[str, int]:
    scratch.mkdir(parents=True, exist_ok=True)
    digest = source_digest(source_uri)
    items = write_manifest(source_uri, digest, zoom, scratch / "manifest.json")

    # return_state=True is what makes partial failure possible. Without it, the
    # first failing item raises and the remaining mapped runs are cancelled.
    states: list[State] = render_tile.map(items, out_dir=scratch, return_state=True)

    paths: list[Optional[str]] = []
    failures: list[TileItem] = []
    for item, state in zip(items, states):
        if state.is_completed():
            paths.append(state.result())
        else:
            paths.append(None)
            failures.append(item)

    if failures:
        get_run_logger().warning(
            "%d/%d tiles failed; first: %s", len(failures), len(items), failures[0].work_key
        )
        dead_letter_tiles(failures)            # captured, not lost

    return {"total": len(items), "ok": len(items) - len(failures), "failed": len(failures)}

Registering the limit is a one-off, and it belongs in deployment rather than in code so it can be changed without a release:

# The limit lives on the server and is shared by every flow that uses the tag.
prefect gcl create scene-reads --limit 8

# Work-pool concurrency is a different bound: how many task runs may exist.
prefect work-pool set-concurrency-limit raster-pool 32

# And a tag limit bounds a class of task across all flows.
prefect concurrency-limit create tile 24
Three limits, three different questionsThe work pool bounds how many task runs execute. The tag bounds how many tasks of this class run across all flows. The named limit bounds how many touch one upstream source.work pool — 32 task runs may executebounds worker capacitytag “tile” — 24 across every flowbounds this class of work globallynamed limit “scene-reads” — 8 concurrent reads of the sourcebounds what the upstream service actually experiencesThe innermost number is the one the source operator cares about, and the one most pipelines never set.
Setting only the outer limit is the common case, and it produces a pipeline that is polite to its own workers and rude to everyone else’s.

The three limits are worth setting deliberately rather than copying, because they answer different questions and their right values rarely coincide. The work-pool limit follows from worker memory and the per-task peak — thirty-two runs on workers that can each hold four concurrent warps. The tag limit follows from how many flows share this class of work: if a nightly mosaic and an on-demand preview both render tiles, the tag is what stops the two from doubling each other. And the named resource limit follows from the endpoint’s published capacity, which is usually the smallest of the three and is the only one an outside party would recognise as a promise.

Where each limit's number comes fromThe work-pool limit comes from worker memory divided by the per-task peak. The tag limit comes from how many flows share the work class. The named resource limit comes from the endpoint's published capacity.worker memory ÷ peaka fact about your machineswork pool = 32flows sharing this worka fact about your platformtag = 24published capacitya fact about someone elsescene-reads = 8Three inputs, threenumbers, no coincidencethat they differ.
Copying one number into all three slots produces a pipeline bounded by whichever fact happened to be copied, which is rarely the one that matters.

Parameter & Option Reference

Parameter Where Default Spatial notes
return_state=True .map() off Required for partial failure. Without it one bad tile cancels the siblings.
tags=["tile"] @task none Ties the task to a global concurrency limit that spans flows.
concurrency("scene-reads") in-task none Bounds concurrent reads of one upstream source, which is what it cares about.
persist_result @task varies On for the manifest, so a resumed run reads it rather than recomputing.
retries @task 2 Per tile. Combined with the fan-out width this is the real load multiplier.
timeout_seconds @task 900 Backstop above the subprocess timeout, so your own error wins.
fan-out guard in-code 20 000 Catches a wrong zoom before the orchestrator schedules a hundred thousand runs.

Verification & Testing

Test the shape of the fan-out, not the tile rendering — that belongs to the task’s own tests.

from prefect.testing.utilities import prefect_test_harness


def test_one_failure_does_not_cancel_the_rest(monkeypatch, tmp_path) -> None:
    calls: list[str] = []

    def flaky(item, out_dir):
        calls.append(item.work_key)
        if item.x == 2101:
            raise RuntimeError("this tile is cursed")
        return str(out_dir / "ok.tif")

    monkeypatch.setattr("mymodule.render_tile.fn", flaky)
    with prefect_test_harness():
        summary = fan_out("fixtures/scene.tif", zoom=12, scratch=tmp_path)

    assert summary["failed"] == 1
    assert summary["ok"] == summary["total"] - 1
    assert len(calls) >= summary["total"], "the siblings must all have been attempted"


def test_rerun_skips_completed_tiles(tmp_path) -> None:
    with prefect_test_harness():
        first = fan_out("fixtures/scene.tif", zoom=12, scratch=tmp_path)
        second = fan_out("fixtures/scene.tif", zoom=12, scratch=tmp_path)
    assert second["ok"] == first["ok"]
    assert rendered_count() == first["ok"], "the second run must not re-render anything"


def test_manifest_is_written_before_mapping(tmp_path) -> None:
    with prefect_test_harness():
        fan_out("fixtures/scene.tif", zoom=12, scratch=tmp_path)
    manifest = json.loads((tmp_path / "manifest.json").read_text())
    assert manifest and {"z", "x", "y", "source_uri"} <= set(manifest[0])

The second test is the one that pays for itself repeatedly. A fan-out without per-item keys looks identical in every observable way until the day it fails halfway and someone re-runs it — at which point it does four hundred tiles of work to produce fourteen.

What a re-run costs, with and without keysWithout per-item keys a re-run after a partial failure renders all four hundred and twelve tiles again. With keys it renders only the fourteen that failed.FIRST RUN — 398 succeed, 14 fail398 renderedRE-RUN WITHOUT KEYS412 rendered again — 96% of it wastedRE-RUN WITH KEYS14 rendered, 398 ledger lookups at ~1 ms each
The ledger lookups are not free, but four hundred of them cost less than a single tile render — which is the whole economic argument for per-item keys.

Common Pitfalls

  • Omitting return_state=True. The default cancels sibling runs on the first failure, so a fan-out of four hundred can be destroyed by one tile with a corrupt source. The flag is the difference between a partial result and none.
  • Relying only on the work-pool limit. It bounds your workers, not the upstream service. Four hundred tile tasks across thirty-two workers still make four hundred requests to one endpoint unless a named limit says otherwise.
  • Returning arrays from mapped tasks. Every mapped run persists its result. Four hundred returned rasters fill the result store, slow every state write, and are retained long after the run. Return paths.
  • Mapping over the bounding box. A diagonal or coastal scene covers a fraction of its own envelope. Intersecting the footprint first commonly halves the fan-out for no loss.
  • No guard on the item count. A zoom off by two is a sixteen-fold fan-out, and by four is two hundred and fifty-six-fold. One comparison before .map() prevents an afternoon of cancelling runs.

Frequently Asked Questions

How many mapped runs can Prefect handle comfortably?

A few thousand per flow run is fine; ten thousand works but the UI becomes slow to load and state writes start to dominate. Past that, map over batches — a task that handles forty tiles has the same parallelism and a fortieth of the bookkeeping. The practical signal is the ratio of summed task duration to run duration: when it drops below about half, the orchestration is the workload.

Where should the concurrency limit be configured?

On the server, not in the code, so it can be changed during an incident without a deploy. prefect gcl create for a named resource limit and prefect concurrency-limit create for a tag both do this. Keeping them in the deployment’s documentation alongside the endpoint’s published capacity is what stops the numbers from becoming folklore.

Should the manifest task be cached?

Yes, keyed on the source digest and zoom. Recomputing a footprint intersection for an unchanged scene is wasted work on every re-run, and caching it means a re-drive reads exactly the same manifest the original run used — which is what makes the two comparable. Use the same content-hashing approach as caching strategies for spatial tasks.

What should happen to failed items?

They should be dead-lettered with their manifest entry, which is already a complete description of the work. That makes the re-drive trivial — the payload is the input — and it keeps the fan-out’s failure handling consistent with the rest of the pipeline. See dead-letter queues for failed geotasks.

Does mapping preserve order?

.map() returns states in the order of the input sequence, so zipping items to states as above is safe. What is not guaranteed is execution order, which is why an expensive-first manifest ordering is only a scheduling hint. Never rely on completion order for correctness.

← Dynamic Task Mapping for Tile Fan-Out

Related in this section