# Building Idempotent Spatial DAGs with Prefect

The failure this page prevents is concrete: a Prefect flow that reprojects and tiles a parcel layer is retried after an object-store timeout, and because its tasks are keyed on nothing more stable than run time, the retry re-runs the spatial join and writes a second, divergent tile set under the same product version. Building idempotent spatial DAGs with Prefect fixes this with cache_key_fn and result persistence keyed on SHA-256(bbox + crs + layer), plus bounded retries with retry_delay_seconds, so a re-run is a cache hit rather than a duplicate write. It is the Prefect-specific implementation of the patterns in Orchestrating Spatial Pipelines in Python, under Spatial Pipeline Orchestration & Observability; the Airflow equivalent is Automating CRS Reprojection in Airflow.

# Prerequisites

Figure — A decision path: compute the cache key, and on a hit return the persisted artifact, otherwise compute, persist, and return.

# Step-by-Step Implementation

Each step is verifiable before you proceed. The controlling idea is that the cache key is computed from the task’s spatial inputs only, so retries and scheduled re-runs converge on one artifact.

# 1. Define the deterministic cache-key function

The key function reads the task arguments and hashes the bounding box, CRS, and layer. It must exclude anything non-deterministic — run id, timestamp, worker name — or caching silently breaks.

# cache.py
import hashlib
from prefect.context import TaskRunContext

def spatial_cache_key(context: TaskRunContext, arguments: dict) -> str:
    parts = f"{arguments['bbox']}|{arguments['crs']}|{arguments['layer']}"
    digest = hashlib.sha256(parts.encode()).hexdigest()
    return f"{arguments['layer']}-{digest}"

Verify: identical inputs must yield an identical key.

python -c "from cache import spatial_cache_key; \
c=None; a={'bbox':'-89.7,39.7,-89.6,39.8','crs':'EPSG:4326','layer':'parcels'}; \
print(spatial_cache_key(c,a)); print(spatial_cache_key(c,a))"

# 2. Bind the key and result persistence to the reproject task

cache_key_fn reuses a prior result for a matching key; persist_result=True writes the artifact to the result store so a fresh process can still hit the cache. cache_expiration bounds staleness.

# tasks.py
from datetime import timedelta
from prefect import task
from cache import spatial_cache_key

@task(cache_key_fn=spatial_cache_key, cache_expiration=timedelta(days=7),
      retries=3, retry_delay_seconds=[10, 30, 90], persist_result=True)
def reproject(bbox: str, crs: str, layer: str) -> str:
    dst = f"s3://{__import__('os').environ['RESULT_BUCKET']}/reprojected/{layer}.parquet"
    run_ogr2ogr(src_layer=layer, dst=dst, t_srs="EPSG:4326")  # canonical CRS
    return dst

Verify: run the task twice; the second run reports a cached state.

prefect flow-run logs <run-id> | grep -i "Finished in state Cached"

# 3. Add the topology gate with zero retries

Validation must fail closed, so it takes retries=0: a retried validation could mask a genuine topology failure and publish invalid geometry.

@task(cache_key_fn=spatial_cache_key, retries=0, persist_result=True)
def validate_topology(bbox: str, crs: str, layer: str, src: str) -> str:
    invalid = count_invalid(src)  # PostGIS ST_IsValid over the reprojected features
    if invalid:
        raise ValueError(f"{invalid} invalid geometries — blocking publication")
    return src

Verify: feed a known-invalid fixture and confirm the flow fails rather than publishing.

prefect flow-run logs <run-id> | grep -i "invalid geometries"

# 4. Compose the flow and persist tiles under the same key

The tile task inherits the same key strategy, so the tile write is idempotent: a retry after a partial write re-resolves to the cached tile set instead of appending a divergent one.

from prefect import flow

@flow(name="prefect-spatial-idempotent")
def pipeline(bbox: str, crs: str, layer: str, version: str):
    r = reproject(bbox, crs, layer)
    v = validate_topology(bbox, crs, layer, r)
    t = tile(bbox, crs, layer, v)          # @task with the same cache_key_fn
    publish(layer, version=version, artifact=t)

Verify: re-run the deployment with identical parameters and confirm no new artifact object appears.

aws s3 ls s3://$RESULT_BUCKET/tiles/parcels/ --recursive | wc -l   # unchanged across runs

# 5. Deploy on a schedule and confirm replay idempotency

A scheduled deployment re-runs on cadence; with deterministic keys, unchanged inputs are cache hits and only new data recomputes.

prefect deploy pipeline.py:pipeline -n nightly --cron "0 2 * * *"
prefect deployment run prefect-spatial-idempotent/nightly

Verify: two consecutive scheduled runs over unchanged data must both resolve every task to Cached.

prefect flow-run logs <run-id> | grep -c "state Cached"   # equals the task count

# Configuration Reference

# Common Failure Modes & Fixes

Every run recomputes; nothing is ever cached. Root cause: a non-deterministic value (run id, datetime.now()) leaked into cache_key_fn, so no two keys match. Fix: hash only bbox, crs, and layer, and assert key stability — python -c "from cache import spatial_cache_key; ..." must print the same digest twice.

Cache hits across genuinely different data. Root cause: the key omits an input that actually changes the output, such as the source snapshot date, so distinct inputs collide. Fix: include the source version or content hash in the digest, then invalidate the stale key with prefect result-store cleanup.

Retry duplicates a tile set. Root cause: the tile task wrote to a path not derived from the key, so the retry appended rather than overwrote. Fix: derive the output path from the layer and key, make the write atomic (aws s3 sync to a key-scoped prefix), and confirm object count is stable across runs.

Validation passes but invalid geometry reaches consumers. Root cause: validate_topology was given retries>0, so a transient failure was retried and eventually masked. Fix: set retries=0 on the gate and re-test against a known-invalid fixture.

Cross-process runs miss the cache. Root cause: persist_result is False, so the result lived only in the first worker’s memory. Fix: set persist_result=True and configure a shared result store block reachable by every worker.

# FAQ

# Related

• Up to the parent: Orchestrating Spatial Pipelines in Python
• Automating CRS Reprojection in Airflow — the same idempotency goal with deterministic output paths
• Topology Validation with Dagster Asset Checks — declaring the topology gate as an asset check
• Spatial Pipeline Orchestration & Observability — the section overview and SLA baselines
• Async Execution for Heavy Spatial Queries — offloading heavy joins off the flow’s critical path