This guide is part of Orchestrating Spatial ETL Pipelines, the reference for turning extract, clean, and load scripts into scheduled, restartable spatial data products.

Orchestrating Spatial Pipelines with Prefect

Prefect earns its place in a geospatial stack the moment the shape of the work stops being knowable ahead of time. An Airflow DAG must declare its tasks when the file is parsed; a Prefect flow decides how wide to fan out only after a task has run and counted the STAC items a query returned or the tiles intersecting a shifting area of interest. For pipelines where “how many downloads tonight?” is answered by a live catalog query rather than a config file, that runtime dynamism is the whole point — and it is exactly where naive Airflow ports fall apart.

The catch is that Prefect’s Python-native ergonomics make it easy to write flows that look correct and quietly misbehave at scale: a .map() over ten thousand tiles that opens ten thousand simultaneous connections to a rate-limited endpoint, retries that hammer a dead host, or GeoDataFrames shuttled between tasks until the result store falls over. This guide maps Prefect’s @flow/@task model onto the five-stage spatial ETL pipeline, then works through the four features that make or break a spatial flow: runtime fan-out, bounded concurrency, retries scoped to the Extract stage, and result persistence to object storage.

The Fan-Out Flow: One Query, N Mapped Downloads, One Reduce

The canonical Prefect spatial flow is a scatter-gather. A single discovery task resolves the work list at runtime, a mapped download task explodes into one task run per item, and a reduce task collates the persisted results into a manifest. The width of the middle band is a runtime value — the flow source never names a number.

Everything downstream depends on getting that middle band right. The list feeding .map() must be computed inside a task so Prefect can plan the fan-out; the download task must persist to object storage rather than return raw bytes; and the simultaneity of the mapped runs must be capped independently of their count. The rest of this guide fills in each piece.

Prerequisites & Environment

Prefect changed its public API meaningfully between the 2.x and 3.x lines, so pin deliberately. The patterns below are written against Prefect 3.x and annotated where 2.14 differs.

• Python 3.9+ for Prefect 2.x, Python 3.9–3.12 for Prefect 3.x.
• Core dependencies — prefect>=3.0 (or prefect==2.14.* if you are pinned to the 2.x line), geopandas>=1.0, pyarrow>=15.0 for GeoParquet I/O, and pystac-client>=0.8 for STAC-driven discovery.
• Optional distributed runner — prefect-dask>=0.3 and dask[distributed]>=2024.1 when a thread pool on one worker is not enough.
• Object storage — an S3, GCS, or Azure bucket the workers can write to, plus a configured result store block for durable task results.

pip install "prefect>=3.0" "geopandas>=1.0" "pyarrow>=15.0" \
            "pystac-client>=0.8" "prefect-dask>=0.3"

Verify the install and start a local API before running any flow:

python -c "import prefect; print(prefect.__version__)"
prefect server start   # local Orion/Prefect API at http://127.0.0.1:4200

Choosing Prefect over the alternatives is a workload decision, not a preference — the orchestrator selection guide for spatial ETL works through when runtime fan-out justifies Prefect over a statically-declared Airflow DAG.

Version / Compatibility Matrix

The 2.x → 3.x transition renamed and relocated several of the exact primitives a spatial fan-out relies on. Pin to one line and read the matrix before copying code between projects.

The single highest-value rule: never mix a Prefect 2.x deployment with 3.x workers. The result serialization format and API schema differ, and a mixed fleet fails at result fetch time with opaque deserialization errors.

Step-by-Step Implementation

Step 1 — Discover: Resolve the Runtime Work List

Discovery is a task, not flow-level code, because Prefect must observe its completion before it can plan the fan-out. Return a plain list of lightweight descriptors — never the payloads themselves — so the object handed to .map() stays small.

import logging
from dataclasses import dataclass

from prefect import task
from pystac_client import Client

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class WorkItem:
    """A single unit of fan-out work — small enough to serialize cheaply."""
    item_id: str
    href: str
    collection: str


@task(retries=2, retry_delay_seconds=5)
def discover_items(
    api_url: str,
    collections: list[str],
    bbox: tuple[float, float, float, float],
    datetime_range: str,
) -> list[WorkItem]:
    """Query a STAC API and return the work list whose length is unknown until now."""
    stac_client = Client.open(api_url)
    search = stac_client.search(
        collections=collections, bbox=bbox, datetime=datetime_range, limit=500
    )
    work = [
        WorkItem(item_id=it.id, href=it.assets["data"].href, collection=it.collection)
        for it in search.items()
    ]
    logger.info("Discovered %d items for fan-out", len(work))
    return work

The discovery step is where a Prefect flow meets the ingestion layer — the same pystac-client catalog sync that streams items page-by-page feeds the list this task returns.

Step 2 — Extract: One Mapped Task per Item, Persisted to Storage

The download task takes a single WorkItem, streams it to object storage, and returns a URI. It carries retries because network failures are transient, and it persists rather than returns bytes so a Dask runner never ships a GeoTIFF across the wire.

from pathlib import Path

import requests
from prefect import task


@task(
    retries=4,
    retry_delay_seconds=[2, 8, 30, 90],  # explicit backoff schedule
    tags=["rate-limited-api"],            # bound by a concurrency limit (Step 4)
    persist_result=True,
)
def download_item(item: WorkItem, output_root: str) -> str:
    """Stream one asset to object storage; return its URI. Idempotent on re-run."""
    dest = Path(output_root) / item.collection / f"{item.item_id}.tif"
    dest.parent.mkdir(parents=True, exist_ok=True)
    if dest.exists():
        return dest.as_uri()  # already present — skip the network entirely

    tmp = dest.with_suffix(".part")
    with requests.get(item.href, stream=True, timeout=60) as resp:
        resp.raise_for_status()
        with tmp.open("wb") as fh:
            for chunk in resp.iter_content(chunk_size=1 << 16):
                fh.write(chunk)
    tmp.rename(dest)  # atomic publish — a crash never leaves a half file
    return dest.as_uri()

Passing a list of retry_delay_seconds gives an explicit backoff curve instead of a flat delay — the fourth retry waits 90 seconds, long enough for a throttled endpoint to recover.

Step 3 — Fan Out with .map() Inside the Flow

The flow calls download_item.map(...) over the runtime list. Arguments that are shared across every mapped run — like output_root — are wrapped in unmapped() so Prefect broadcasts them instead of iterating them character by character.

from prefect import flow, unmapped
from prefect.futures import wait


@flow(name="stac-fan-out")
def stac_fan_out_flow(
    api_url: str,
    collections: list[str],
    bbox: tuple[float, float, float, float],
    datetime_range: str,
    output_root: str,
) -> list[str]:
    """Discover N items, download them concurrently, reduce to a manifest."""
    items = discover_items(api_url, collections, bbox, datetime_range)

    # One task run per item; width == len(items), resolved at runtime.
    futures = download_item.map(items, output_root=unmapped(output_root))

    wait(futures)  # block until every mapped run settles
    uris = [f.result() for f in futures]
    return build_manifest(uris)

In Prefect 2.14 the same fan-out works, but unmapped is imported from prefect.utilities.annotations and wait is spelled prefect.futures.wait only from 3.x — under 2.14 you iterate the returned futures and call .result() directly, which blocks per future.

Step 4 — Bound Concurrency for Rate-Limited Endpoints

A ten-thousand-wide .map() will try to run ten thousand task runs as fast as the task runner allows. A tag-scoped global concurrency limit caps the simultaneous runs regardless of fan-out width, keeping the download rate inside the API’s envelope.

# Register once: at most 6 tasks tagged 'rate-limited-api' run at a time.
prefect concurrency-limit create rate-limited-api 6

Because download_item is decorated with tags=["rate-limited-api"], Prefect admits only six of its runs concurrently while the rest queue — no change to the flow code is needed. The full treatment of bounded mapping and per-tile error isolation lives in parallel tile downloads with Prefect task mapping.

Step 5 — Choose a Task Runner

The default ConcurrentTaskRunner uses a thread pool — correct for I/O-bound download fan-out, where threads wait on sockets. Switch to the Dask runner only when tasks are CPU-bound (reprojection, raster warping) and must spread across machines.

from prefect import flow
from prefect_dask import DaskTaskRunner

# Threads: right for download-heavy, I/O-bound fan-out (the default).
@flow(name="io-bound-fan-out")  # ConcurrentTaskRunner is implicit in 3.x
def io_flow(): ...

# Dask: right for CPU-bound reprojection across a cluster.
@flow(task_runner=DaskTaskRunner(cluster_kwargs={"n_workers": 4, "threads_per_worker": 2}))
def cpu_flow(): ...

A common mistake is reaching for Dask on a download-bound flow. Threads already overlap I/O waits; Dask adds serialization overhead and a scheduler for no gain unless the bottleneck is CPU.

Advanced Patterns & Edge Cases

Content-Addressed Caching of Expensive Tasks

Reprojection and re-download are the two most expensive spatial operations, and both are pure functions of their inputs — an unchanged tile reprojected to the same CRS yields the same bytes. Prefect’s caching skips recompute when a cache key repeats. The subtlety for spatial work is building a cache key from the inputs that actually matter — a tile bbox, target CRS, and source ETag — rather than a whole GeoDataFrame. That custom key design is the subject of caching geospatial task results in Prefect.

Partial Failure Without Killing the Flow

By default a failed mapped run can fail the whole flow. For a thousand-tile download, one corrupt scene must not discard nine hundred ninety-nine good ones. Returning states rather than results (return_state=True) and filtering the completed set lets the flow harvest every success and quarantine the failures — the per-tile isolation pattern in parallel tile downloads with Prefect task mapping.

Deployments and Schedules

A flow becomes a production pipeline when it is deployed with a schedule and served by a worker pool. In Prefect 3.x, flow.serve() or a prefect.yaml deployment binds the flow to a cron or interval schedule and a work pool:

if __name__ == "__main__":
    stac_fan_out_flow.serve(
        name="nightly-stac-sync",
        cron="0 2 * * *",                       # 02:00 daily
        parameters={"collections": ["sentinel-2-l2a"]},
    )

The schedule sets the cadence; the flow’s discovery task still decides the fan-out width fresh on every run, so a night with more new scenes simply maps wider.

Performance Optimization

The dominant cost in a spatial fan-out is rarely CPU — it is the serialization of results and the round-trips to the result store. Persisting a GeoDataFrame return value forces Prefect to pickle it; persisting a URI to a GeoParquet file the task already wrote costs a few bytes. The difference is measurable on wide fan-outs:

import time

import geopandas as gpd
from prefect import flow, task


@task(persist_result=True)
def return_gdf(path: str) -> gpd.GeoDataFrame:
    return gpd.read_parquet(path)          # result store pickles the whole frame


@task(persist_result=True)
def return_uri(path: str) -> str:
    # transform writes its own GeoParquet, returns only the pointer
    out = path.replace(".parquet", ".out.parquet")
    gpd.read_parquet(path).to_parquet(out)
    return out                              # result store holds ~80 bytes


@flow
def benchmark(paths: list[str]) -> None:
    t0 = time.perf_counter()
    return_gdf.map(paths)
    t1 = time.perf_counter()
    return_uri.map(paths)
    t2 = time.perf_counter()
    print(f"return GeoDataFrame: {t1 - t0:.1f}s | return URI: {t2 - t1:.1f}s")

On a 500-tile fan-out with ~40 MB frames, returning URIs instead of GeoDataFrames typically cuts wall-clock time by 3–5x and removes result-store size pressure entirely, because the mapped runs no longer contend on pickling large payloads.

Integration into ETL Pipelines

A Prefect flow slots into the five-stage spatial model cleanly: discover_items is Discover, the mapped download_item is Extract, a mapped transform task is Transform, a validation task splits passing and failing records for a dead-letter store, and a final upsert task is Load. Keep every inter-task handoff a URI to GeoParquet in object storage — that is what makes a Dask runner viable and what lets a retried task pick up from the last persisted result instead of recomputing.

For dead-lettering, have the reduce step separate the completed manifest from the failed items and write failures to a quarantine prefix, then alert on quarantine volume rather than on individual task failures. Emit structured logs keyed by item_id at each task boundary so an operator can trace a single tile through Discover, Extract, and Load without reading a stack trace. When the fan-out feeds a warehouse, the Load task’s upsert should key on the same stable item_id used during discovery, so a re-run overwrites rather than appends.

Troubleshooting Reference

Related

• Caching Geospatial Task Results in Prefect — content-addressed cache keys over tile bbox, CRS, and source ETag so unchanged inputs skip recompute
• Parallel Tile Downloads with Prefect Task Mapping — bounded .map() concurrency and per-tile error isolation for partial failures
• Orchestrating Spatial ETL Pipelines — the five-stage reference model and where Prefect fits among Airflow and Dagster
• Choosing an Orchestrator for Spatial ETL — when runtime fan-out justifies Prefect over a static DAG
• Syncing STAC Catalogs with pystac-client — the discovery layer that produces the runtime item list a Prefect flow maps over