Parameterizing Prefect Flows for Multi-Tile Runs

A flow that processes one tile becomes a flow that processes three hundred by taking the tile list as a typed parameter and fanning out:

from prefect import flow

@flow(name="tile-indices")
def process_tiles(tiles: list[str], start: str, end: str, indices: list[str] = ["ndvi"]):
    for tile in tiles:
        process_one_tile(tile=tile, start=start, end=end, indices=indices)

Parameterisation is what turns the scheduled flows in Orchestrating Pipelines with Prefect into something that can be re-run for one tile, one date, or one failed subset.

Why This Arises in Remote Sensing Workflows

Raster pipelines are naturally parameterised: a tile, a date window, a set of indices, an output prefix. When those values are hard-coded, every variation becomes a code change, and re-running a single failed tile means editing a file, committing, and redeploying.

The operational cost shows up on the bad days. A run of three hundred tiles fails on eleven of them because of transient storage errors. With a parameterised flow, the recovery is one run with eleven tile identifiers. Without it, the choice is between re-running everything — expensive, and re-writing two hundred and eighty-nine good outputs — and hand-editing a script under time pressure.

Parameterisation also makes the run legible. A UI showing three hundred runs called process-tiles is useless; one showing ndvi-36MYF-2023-07 next to ndvi-36NYG-2023-07 lets an operator see at a glance which tiles are slow, which failed, and which have not started.

What parameterisation buys operationally A hard-coded flow has one shape and any variation requires a code change. A parameterised flow takes tiles, dates and options as inputs, fans out one subflow per tile, and can be re-run for an arbitrary subset — which is what makes recovering eleven failed tiles a one-line operation. One deployment, many runs parameters tiles · start · end indices · overwrite parent flow validates, fans out, summarises subflow: 36MYF own retries, own state subflow: 36NYG own retries, own state subflow: 37NBA — failed re-run with tiles=["37NBA"] Recovery is a new run with a different parameter value — no code change, no redeploy.

Environment & Setup

Package Version Why
prefect ≥2.14 Flows, subflows, .map, tags, run-name templates
pydantic ≥2.0 Typed parameter validation
pystac-client ≥0.7 Resolving the items each tile needs
rioxarray ≥0.15 The per-tile work itself 
pip install "prefect>=2.14" "pydantic>=2.0" "pystac-client>=0.7" "rioxarray>=0.15"

Complete Working Example

A parent flow that validates parameters, fans out one subflow per tile with bounded concurrency, and returns a summary that names the failures.

Subflow per tile versus mapped task Subflows give each tile its own state, retries, logs and run name, which is what makes a 300-tile run debuggable. Mapped tasks are lighter and better suited to thousands of small uniform units where per-unit observability is not worth the scheduling overhead. Choosing the fan-out mechanism property subflow per tile mapped task per-unit retries yes, flow-level task-level only own run name in the UI yes no scheduling overhead higher lower good for minutes per tile seconds per unit debugging one failure open its run search the parent run log A useful rule: subflow if a unit takes minutes, mapped task if it takes seconds. 
from datetime import date

from prefect import flow, task, get_run_logger
from prefect.task_runners import ConcurrentTaskRunner
from pydantic import BaseModel, Field, field_validator


class RunConfig(BaseModel):
    """Typed parameters — validation happens before any work starts."""
    tiles: list[str] = Field(..., min_length=1)
    start: date
    end: date
    indices: list[str] = Field(default_factory=lambda: ["ndvi"])
    max_cloud: int = Field(20, ge=0, le=100)
    overwrite: bool = False

    @field_validator("tiles")
    @classmethod
    def tiles_look_like_mgrs(cls, v: list[str]) -> list[str]:
        bad = [t for t in v if not (4 <= len(t) <= 6)]
        if bad:
            raise ValueError(f"these do not look like MGRS tiles: {bad}")
        return v

    @field_validator("end")
    @classmethod
    def end_after_start(cls, v: date, info) -> date:
        if "start" in info.data and v < info.data["start"]:
            raise ValueError("end is before start")
        return v


@task(retries=3, retry_delay_seconds=[10, 30, 90], tags=["stac-read"])
def search_tile(tile: str, start: date, end: date, max_cloud: int) -> list[str]:
    """Item ids for one tile and window. Tagged so its concurrency can be capped."""
    ...
    return item_ids


@task(retries=2, retry_delay_seconds=[20, 60], tags=["raster-io"])
def compute_index(item_id: str, index: str, overwrite: bool) -> str:
    """One item, one index, deterministic output path."""
    ...
    return output_uri


@flow(name="tile-index", flow_run_name="{tile}-{index_label}")
def process_tile(tile: str, cfg: RunConfig, index_label: str = "ndvi") -> dict:
    """Everything for one tile. Its own state, retries and logs."""
    logger = get_run_logger()
    item_ids = search_tile(tile, cfg.start, cfg.end, cfg.max_cloud)
    logger.info("tile %s: %d items in window", tile, len(item_ids))

    futures = [compute_index.submit(i, ix, cfg.overwrite)
               for i in item_ids for ix in cfg.indices]
    written, failed = [], []
    for fut in futures:
        try:
            written.append(fut.result())
        except Exception as exc:
            failed.append(repr(exc))

    return {"tile": tile, "items": len(item_ids), "written": len(written),
            "failed": len(failed), "errors": failed[:3]}


@flow(name="multi-tile-indices", task_runner=ConcurrentTaskRunner())
def process_tiles(
    tiles: list[str],
    start: date,
    end: date,
    indices: list[str] | None = None,
    max_cloud: int = 20,
    overwrite: bool = False,
) -> dict:
    """Parent flow: validate once, fan out per tile, summarise."""
    cfg = RunConfig(tiles=tiles, start=start, end=end,
                    indices=indices or ["ndvi"], max_cloud=max_cloud, overwrite=overwrite)
    logger = get_run_logger()
    logger.info("processing %d tile(s) from %s to %s", len(cfg.tiles), cfg.start, cfg.end)

    results = []
    for tile in cfg.tiles:
        # A subflow per tile: failures are isolated and visible per tile in the UI
        results.append(process_tile(tile=tile, cfg=cfg, index_label="-".join(cfg.indices)))

    failed_tiles = [r["tile"] for r in results if r["failed"]]
    summary = {
        "tiles": len(results),
        "items": sum(r["items"] for r in results),
        "written": sum(r["written"] for r in results),
        "failed_tiles": failed_tiles,
    }
    if failed_tiles:
        logger.warning("re-run with tiles=%s", failed_tiles)
    return summary


if __name__ == "__main__":
    print(process_tiles(tiles=["36MYF", "36NYG"], start=date(2023, 7, 1), end=date(2023, 7, 31)))

Two details make this usable at scale. flow_run_name="{tile}-{index_label}" turns the run list into something an operator can scan. And returning failed_tiles from the parent means the recovery command writes itself — the log line literally contains the parameter for the retry run.

Variant Patterns

1. Bounding concurrency with tags

Why a 300-tile run needs a concurrency cap Without a limit, three hundred tiles each submitting reads produce thousands of simultaneous requests, and the storage backend responds with throttling errors that then consume the retry budget. A tag-based limit caps simultaneous reads regardless of how many tiles are in flight, so throughput is steady and no retries are wasted. Requests in flight during a 300-tile run unbounded — spikes, throttling, retries tag limit 16 — steady, no throttling start end of run prefect concurrency-limit create raster-io 16 The cap applies across every run, so two overlapping jobs cannot exceed it either. 
prefect concurrency-limit create raster-io 16
prefect concurrency-limit create stac-read 8

Because the tasks above already carry tags=["raster-io"] and tags=["stac-read"], no code change is needed to apply the caps — and the caps hold across concurrent runs, which per-flow settings do not. This composes with the backoff policy in Retrying Failed Raster Tasks in a Prefect Pipeline: the cap prevents most throttling, and the backoff handles what gets through.

2. Mapped tasks instead of subflows

For small, uniform units, mapping is lighter than a subflow per tile.

@flow
def process_tiles_mapped(tiles: list[str], start: date, end: date):
    futures = compute_tile.map(tile=tiles, start=start, end=end)
    return [f.result() for f in futures]

The trade is observability: mapped task runs are visible but they do not have their own flow-level state, retries or logs, which makes debugging a specific tile harder. As a rule, if a tile takes minutes, use a subflow; if it takes seconds, map it.

3. Deriving the tile list rather than passing it

Sometimes the caller knows an area, not a tile list. Resolving the list inside the flow keeps the interface simple and the resolution auditable.

@task
def tiles_for_aoi(bbox: tuple[float, float, float, float]) -> list[str]:
    """MGRS tiles intersecting a bounding box, from the catalogue's own footprints."""
    items = catalog.search(collections=["sentinel-2-l2a"], bbox=bbox, limit=100).items()
    return sorted({i.properties["s2:mgrs_tile"] for i in items})

Log the resolved list. A run that silently processes a different set of tiles than expected is much harder to debug than one whose first log line names them.

Making a Large Run Observable

Three practices make the difference between a run you can supervise and one you can only wait for.

Name every run from its parameters. Both the parent and the subflows should carry the tile and the window in their names, so the UI is a work list rather than a wall of identical entries.

Return structured summaries, not None. A parent flow that returns counts and failed identifiers gives you a machine-readable artefact for the retry and a human-readable one for the status report. It also makes the flow testable without a scheduler.

Log the decision points, not the progress. “tile 36MYF: 14 items in window, 3 skipped as existing” is worth logging; a line per item is noise that hides it. The per-item detail belongs in the status table described in Batch Computing Indices Across a STAC Collection.

Common Errors

The run fails immediately with a validation error

Working as intended: the typed parameters rejected the input before any compute was spent. Fix the parameter rather than loosening the validator.

Subflow runs do not appear in the UI

They were called from inside a task rather than from the flow. Subflows must be invoked from flow context; a flow called inside a task runs but is not tracked as a subflow.

A large run overwhelms the storage backend

The concurrency limits are not applied, or the tasks are missing their tags. Check that the tag names in the code match the limits registered on the server.

Frequently Asked Questions

Q: Subflow per tile or one mapped task? Subflows when each tile is substantial and you want per-tile retries, states and logs that can be inspected independently. Mapped tasks when tiles are small and uniform, because thousands of subflows add scheduling overhead and clutter the UI.

Q: How do I stop a big run from throttling the bucket? Use a tag-based concurrency limit. Tagging the read tasks and capping that tag at, say, sixteen bounds simultaneous requests regardless of how many tiles are in flight.

Q: Can I re-run only the failed tiles? Yes, if the flow is parameterised by tile and each tile writes a deterministic output. Collect the failed tile identifiers from the run and pass them as the tiles parameter of a new run.