This guide is part of Orchestrating Spatial Pipelines with Prefect, within the broader Orchestrating Spatial ETL Pipelines reference.

Parallel Tile Downloads with Prefect Task Mapping

The Problem: Unbounded Fan-Out Either Rate-Limits You or Fails on One Bad Tile

Downloading a few thousand map tiles or satellite scenes is embarrassingly parallel — until it meets a rate-limited API and a single corrupt tile. Prefect’s .map() makes the fan-out trivial to express, but the two failure modes it invites are exactly the ones that take down production spatial flows:

• Fan-out width is not a rate limit. A .map() over 5,000 tile bboxes schedules 5,000 task runs; with no bound, the task runner admits as many as it can and the download endpoint answers with a wall of HTTP 429s, poisoning even the requests that would have succeeded.
• One failed tile fails everything. By default a raised exception in a mapped run propagates, and the flow aborts — discarding thousands of tiles that downloaded cleanly and forcing a full re-run.
• Retries amplify the flood. Retrying every 429 without a concurrency ceiling turns a throttle event into a self-reinforcing storm, because the retries pile onto the same saturated endpoint.
• Silent partial coverage. Without explicit success/failure partitioning, a flow can appear to finish while a subset of tiles never landed, leaving gaps in a mosaic that surface only during analysis.

The fix is two independent controls: a global concurrency limit that decouples fan-out width from simultaneous execution, and per-tile error isolation that turns a failure into a quarantined record instead of an aborted flow.

Version and Environment Compatibility

pip install "prefect>=3.0" "geopandas>=1.0" requests

The diagram below shows how the concurrency limit gates a wide fan-out and where failures branch off to the dead-letter set.

The Recipe: A Bounded, Fault-Isolated Mapped Download

The task downloads one tile bbox and is tagged so a global concurrency limit throttles it. The flow maps it over the runtime tile list with return_state=True, then partitions the settled states into completed and failed. Nothing raises out of the fan-out.

import logging
from dataclasses import dataclass
from pathlib import Path

import requests
from prefect import flow, task, unmapped
from prefect.futures import wait
from prefect.states import State

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class Tile:
    """One unit of download work, keyed by a stable tile id."""
    tile_id: str
    bbox: tuple[float, float, float, float]
    url: str


@task(
    tags=["tile-api"],                      # bound by the concurrency limit below
    retries=3,
    retry_delay_seconds=[5, 20, 60],        # backoff so retries do not amplify a 429 storm
)
def download_tile(tile: Tile, output_root: str) -> str:
    """Download one tile to object storage; return its URI. Idempotent on re-run."""
    dest = Path(output_root) / f"{tile.tile_id}.tif"
    dest.parent.mkdir(parents=True, exist_ok=True)
    if dest.exists():
        return dest.as_uri()                # already present — skip network + a slot

    tmp = dest.with_suffix(".part")
    with requests.get(tile.url, params={"bbox": ",".join(map(str, tile.bbox))},
                      stream=True, timeout=60) as resp:
        resp.raise_for_status()             # a 4xx/5xx raises → this run's state is Failed
        with tmp.open("wb") as fh:
            for chunk in resp.iter_content(chunk_size=1 << 16):
                fh.write(chunk)
    tmp.rename(dest)                         # atomic publish
    return dest.as_uri()


@flow(name="bounded-tile-download")
def download_tiles_flow(tiles: list[Tile], output_root: str) -> dict:
    """Map the download over every tile, bounded by concurrency, isolating failures."""
    # return_state=True → each element is a State, not a raised result.
    states: list[State] = download_tile.map(
        tiles, output_root=unmapped(output_root), return_state=True
    )
    wait(states)

    completed: list[str] = []
    dead_letter: list[str] = []
    for tile, state in zip(tiles, states):
        if state.is_completed():
            completed.append(state.result())          # the returned URI
        else:
            logger.warning("Tile %s failed after retries: %s", tile.tile_id, state.type)
            dead_letter.append(tile.tile_id)           # quarantine, do not raise

    logger.info("Downloaded %d/%d tiles; %d dead-lettered",
                len(completed), len(tiles), len(dead_letter))
    return {"completed": completed, "dead_letter": dead_letter}

Register the concurrency limit once, sized to the API’s rate envelope rather than the worker count:

# At most 8 'tile-api' task runs execute simultaneously, however wide .map() gets.
prefect concurrency-limit create tile-api 8

Key Implementation Notes

• return_state=True is what converts a failure into data. Without it, the first tile that exhausts its retries raises out of .map() and aborts the flow. With it, each run settles into a State you inspect after wait(), so the flow decides what a failure means instead of the exception dictating it.
• The concurrency limit throttles execution, not scheduling. All N runs are scheduled immediately; the limit gates how many leave the queue. This is why a 5,000-tile .map() with a limit of 8 is safe — the download rate is governed by K, not N.
• Size K to the slowest external dependency. The rate-limited endpoint, not the number of workers, is the real constraint. A worker pool can exceed K safely because the concurrency limit is the throttle; setting K above the API’s requests-per-second envelope reintroduces the 429 storm.
• Retries carry a backoff curve, not a flat delay. A retry_delay_seconds list spaces retries so a transient throttle recovers between attempts, rather than three immediate retries hammering an already-saturated endpoint.
• A cache hit or an existing file never consumes a slot on the network path. The dest.exists() short-circuit returns before the request, so a mostly-complete re-run clears fast and only the genuinely-missing tiles contend for the limited slots.
• zip(tiles, states) preserves the tile-to-state mapping. .map() returns states in input order, so pairing each state with its originating tile gives you the tile id to write into the dead-letter list — essential for a targeted re-run of only the failures.

Troubleshooting Partial-Failure and Rate-Limit Issues

Integration Note

Bounded mapping is the workhorse of the Extract stage in orchestrating spatial pipelines with Prefect: the discovery task resolves the tile list, this recipe downloads it under a rate cap, and the reduce step consumes the completed/dead_letter split to build a manifest and alert on quarantine volume. It pairs directly with caching geospatial task results in Prefect — because cache hits return before the task body runs, a cached tile never consumes a concurrency slot, so a mostly-cached nightly run finishes in seconds while only the new tiles queue against the download limit.

Related

• Orchestrating Spatial Pipelines with Prefect — the flow structure and runtime fan-out this bounded, fault-isolated download plugs into
• Caching Geospatial Task Results in Prefect — content-addressed cache keys that let a mostly-cached fan-out skip the concurrency queue entirely
• Orchestrating Spatial ETL Pipelines — the five-stage reference model, dead-lettering, and concurrency-pool sizing for rate-limited endpoints