Orchestrating Pipelines with Prefect

A raster processing script that works once on your laptop rarely survives contact with production. It needs to run on a schedule, recover from a transient S3 timeout without re-doing hours of work, run scenes in parallel, and leave an audit trail when something fails at 3 a.m. Prefect 2.x turns an ordinary Python function into an observable, retryable workflow with a few decorators, which makes it the natural control plane for Earth observation pipelines. For where orchestration sits relative to compute scaling and cost control, see the parent section on Cloud Execution & Orchestration.

This guide is aimed at remote sensing analysts and data engineers who already have working rasterio or xarray code and now need to operationalize it. We decompose a canonical STAC → download → process pipeline into @task and @flow units, fan the work out with .submit, add retries and caching, attach a cron schedule through a deployment, and cap shared-resource usage with concurrency limits. For raw compute throughput — as opposed to orchestration — pair these patterns with Scaling Raster Processing with Dask; Prefect schedules and supervises the run, Dask parallelizes the array math inside it.

---

Orchestration architecture overview

The diagram below shows how Prefect wraps a linear raster pipeline. The scheduler triggers a flow run; the flow calls tasks; tasks are individually retried, cached, and mapped across scenes; the flow reports state back to the API and UI.

The essential mental model: a @flow is the orchestration boundary — one run, one row in the UI — and each @task is an independently supervised unit of work inside it. Retries, caching, and parallel mapping are properties of tasks; scheduling and overall run state are properties of the flow.

---

Prerequisites

Install Prefect alongside the raster stack. Pin the Prefect 2.x major line; the 3.x API renamed several primitives and the decorators behave differently.

pip install "prefect>=2.14,<3" "pystac-client>=0.7" \
            "rasterio>=1.3" "rioxarray>=0.15" "xarray>=2023.1" "numpy>=1.24"

Conceptual prerequisites: you should already know how to build a STAC search — the search step here reuses the patterns in Querying STAC Catalogs Programmatically — and be comfortable with basic index math. Start a local API and UI with prefect server start, or point at Prefect Cloud by setting PREFECT_API_URL. Nothing below requires Cloud; every example runs against a local server.

---

Step-by-step workflow

Step 1 — Wrap the STAC search as a @task

The first unit of work is the catalog query. Wrapping it in @task gives it its own logger, retry policy, and a place in the run graph. Return plain, serializable data (asset URLs and IDs) rather than live client objects so results can be cached and passed between tasks cleanly.

from datetime import date, timedelta
from prefect import task, get_run_logger
from pystac_client import Client

STAC_API = "https://earth-search.aws.element84.com/v1"

@task(name="search-stac", retries=2, retry_delay_seconds=10)
def search_stac(bbox: tuple[float, float, float, float], days: int = 7) -> list[dict]:
    """Return a list of {id, href} dicts for recent Sentinel-2 L2A red-band assets."""
    logger = get_run_logger()
    end = date.today()
    start = end - timedelta(days=days)
    client = Client.open(STAC_API)
    search = client.search(
        collections=["sentinel-2-l2a"],
        bbox=bbox,
        datetime=f"{start.isoformat()}/{end.isoformat()}",
        query={"eo:cloud_cover": {"lt": 20}},
    )
    items = [
        {"id": item.id, "href": item.assets["red"].href}
        for item in search.items()
    ]
    logger.info("STAC search returned %d items", len(items))
    return items

get_run_logger() is the correct way to log from inside Prefect — it routes messages to both stdout and the run record in the UI, so a scheduled run leaves a searchable trail. Never use bare print(); those lines are lost once the process exits.

Step 2 — Fan out downloads with .submit for parallel mapping

Calling a task normally (download_asset(item)) blocks until it returns. Calling download_asset.submit(item) returns a PrefectFuture immediately and hands execution to the flow’s task runner. Submitting across a list fans the work out so scenes download concurrently instead of one at a time.

from pathlib import Path
import rasterio
from prefect import task, get_run_logger

@task(name="download-asset", retries=3, retry_delay_seconds=[10, 30, 90])
def download_asset(item: dict, out_dir: str) -> str:
    """Stream one COG to local disk and return its path."""
    logger = get_run_logger()
    dest = Path(out_dir) / f"{item['id']}.tif"
    with rasterio.open(item["href"]) as src:
        profile = src.profile
        data = src.read()
    with rasterio.open(dest, "w", **profile) as dst:
        dst.write(data)
    logger.info("Downloaded %s (%d bands)", item["id"], data.shape[0])
    return str(dest)

Inside the flow you then map with .submit:

futures = [download_asset.submit(item, out_dir) for item in items]
paths = [f.result() for f in futures]   # blocks until all complete

With the default ConcurrentTaskRunner, these run in a thread pool — ideal because downloads are I/O-bound and spend most of their time waiting on the network. For CPU-bound array math, swap in DaskTaskRunner so submitted tasks execute across Dask workers.

Step 3 — Add retries and task result caching

Transient failures — a dropped S3 connection, a 503 from the STAC API — should not sink an entire run. retries plus a list-valued retry_delay_seconds gives exponential-style backoff per attempt. Caching goes further: a cache_key_fn lets a re-run skip a scene that was already processed, which is what makes the pipeline idempotent and resumable.

from datetime import timedelta
from prefect import task
from prefect.tasks import task_input_hash

@task(
    name="process-scene",
    retries=2,
    retry_delay_seconds=30,
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(days=7),
)
def process_scene(scene_path: str, out_dir: str) -> str:
    """Compute an index and write a COG. Cached by input hash for 7 days."""
    ...
    return output_path

task_input_hash builds the cache key from the task’s arguments, so calling process_scene again with the same scene_path returns the stored result without re-executing. For the narrow, resilience-focused view of retries — backoff schedules, timeouts, and retrying only transient errors — see Retrying Failed Raster Tasks in Prefect.

Step 4 — Compose the @flow and emit run logs

The @flow ties the tasks together. It owns the task runner, the run-level logger, and the overall state that appears in the UI. Keep orchestration logic — mapping, gathering futures, branching — inside the flow, and keep heavy I/O and compute inside tasks.

from prefect import flow, get_run_logger
from prefect.task_runners import ConcurrentTaskRunner

@flow(name="s2-index-pipeline", task_runner=ConcurrentTaskRunner())
def s2_pipeline(
    bbox: tuple[float, float, float, float],
    out_dir: str = "./out",
    days: int = 7,
) -> list[str]:
    logger = get_run_logger()
    items = search_stac(bbox, days=days)
    if not items:
        logger.warning("No scenes matched; nothing to process.")
        return []

    dl_futures = [download_asset.submit(item, out_dir) for item in items]
    proc_futures = [
        process_scene.submit(f.result(), out_dir) for f in dl_futures
    ]
    outputs = [f.result() for f in proc_futures]
    logger.info("Pipeline produced %d index rasters", len(outputs))
    return outputs

if __name__ == "__main__":
    s2_pipeline(bbox=(11.0, 46.0, 11.5, 46.5))

Because download_asset returns paths and process_scene consumes them, Prefect infers the dependency and schedules processing only after the matching download completes — no manual sequencing required.

Step 5 — Build a deployment with a cron schedule

A flow you have to launch by hand is not orchestrated. A deployment registers the flow with the API and attaches a schedule so a worker picks it up automatically. CronSchedule uses standard cron syntax; here the pipeline runs every day at 03:00 UTC.

from prefect.deployments import Deployment
from prefect.server.schemas.schedules import CronSchedule
from pipeline import s2_pipeline

deployment = Deployment.build_from_flow(
    flow=s2_pipeline,
    name="nightly-s2-index",
    schedule=CronSchedule(cron="0 3 * * *", timezone="UTC"),
    parameters={"bbox": [11.0, 46.0, 11.5, 46.5], "days": 2},
    work_queue_name="raster",
)

if __name__ == "__main__":
    deployment.apply()

Run python deploy.py to register it, then start a worker with prefect agent start -q raster. The scheduler now creates a run each night; the worker executes it. Use IntervalSchedule(interval=timedelta(hours=6)) instead if you want fixed spacing rather than wall-clock times. A complete, end-to-end scheduled downloader — with manifest-based deduplication — is walked through in Scheduling Sentinel-2 Downloads with Prefect.

Step 6 — Apply concurrency limits to protect shared resources

Fanning out with .submit is powerful enough to overwhelm a rate-limited STAC API or a database connection pool. A tag-based concurrency limit caps how many task runs carrying a given tag execute at once, across every flow run in the deployment.

# Allow at most 4 concurrent tasks tagged "s3-download"
prefect concurrency-limit create s3-download 4

Tag the relevant tasks so the limit applies:

@task(name="download-asset", tags=["s3-download"], retries=3)
def download_asset(item: dict, out_dir: str) -> str:
    ...

Prefect now queues submitted s3-download tasks beyond the fourth until a slot frees up. This decouples parallelism from politeness: you can submit a hundred scenes while guaranteeing the archive never sees more than four simultaneous connections.

---

Parameter reference

---

Verification & testing

Test flows and tasks like ordinary functions. Calling a @flow directly executes it synchronously and returns the result, so a unit test needs no server:

from pipeline import s2_pipeline

def test_pipeline_handles_empty_search(monkeypatch):
    # Force the search task to return no items
    monkeypatch.setattr("pipeline.search_stac", lambda *a, **k: [])
    result = s2_pipeline.fn(bbox=(0, 0, 0.1, 0.1))
    assert result == []

Use .fn to call the undecorated function when you want to bypass task-run tracking entirely. To confirm the schedule is registered, inspect it from the CLI:

prefect deployment ls          # confirm nightly-s2-index is listed
prefect deployment inspect s2-index-pipeline/nightly-s2-index

The inspect output includes the resolved next run times — verify they fall where the cron expression predicts before trusting the schedule in production.

---

Troubleshooting

RuntimeError: get_run_logger() called outside of a flow or task run — you invoked a task’s function directly (via .fn) or imported logging into module scope. Only call get_run_logger() inside a @task or @flow body executed by Prefect.

Submitted tasks run sequentially, not in parallel — the flow is using SequentialTaskRunner, or you called .result() inside the submission loop, which blocks each iteration. Build the full list of futures first, then gather results in a second pass.

Cached results are stale after fixing a bug — task_input_hash keys only on arguments, not on the function body. After changing task logic, bump a version argument or set a new cache_key_fn, or clear the result store, so old results are not reused.

Scheduled runs never start — a deployment with a schedule still needs a running worker polling its work_queue_name. Start one with prefect agent start -q raster and confirm the queue name matches the deployment.

Crash detected! Execution was interrupted by an unexpected exception — the flow process was killed (often an OOM from loading whole scenes into memory). Reduce per-task memory with windowed reads and cap fan-out with a concurrency limit; see the memory patterns in Optimizing Rasterio Window Reads for Memory Efficiency.

---

FAQ

What is the difference between a Prefect @flow and a @task?

A @flow is the orchestration boundary — it defines a run, owns retries at the run level, and appears as one entry in the Prefect UI. A @task is a single unit of work inside a flow, such as one download or one NDVI calculation; tasks are individually retried, cached, and mapped for parallelism. Only functions called from within a flow are tracked as task runs.

How do I run Prefect tasks in parallel over many Sentinel-2 scenes?

Call task.submit() inside the flow to return a PrefectFuture instead of a blocking result, then submit across a list of items to fan out the work. With a concurrent or Dask task runner configured on the flow, submitted tasks execute in parallel. Collect results by calling .result() on each future or by passing the futures as dependencies to a downstream task.

Does task caching persist across separate Prefect flow runs?

Yes. When you set a cache_key_fn on a @task, Prefect stores the result keyed by that value and reuses it on any later run whose inputs produce the same key, until cache_expiration elapses. This lets a re-run skip scenes that were already downloaded and processed, which is the mechanism behind idempotent, resumable raster pipelines.

---

Related

• Cloud Execution & Orchestration — parent section covering scheduling, distributed compute, and cost control across the pipeline lifecycle
• Scaling Raster Processing with Dask — parallelize the array math inside your Prefect tasks with a Dask task runner
• Querying STAC Catalogs Programmatically — build the STAC search that feeds the download stage
• Scheduling Sentinel-2 Downloads with Prefect — a full cron-scheduled downloader with manifest deduplication
• Retrying Failed Raster Tasks in Prefect — resilience patterns for transient S3 and GDAL errors