Writing Idempotent Airflow DAGs for Geospatial Refresh

This page solves one exact problem: making each task in a geospatial Airflow refresh safe to re-run, so a retry or a backfill produces byte-identical output instead of duplicated rows or half-written files.

Idempotency is the property that turns a fragile pipeline into a resumable one. When the routing engine dies halfway through a national run, you want to re-run the DAG and have finished regions skipped, the failed region recomputed, and nothing written twice. That guarantee does not come for free — an INSERT-only load or a direct-to-final-path write breaks it silently. This page shows the three techniques that make geospatial tasks idempotent (deterministic keys, skip-if-exists, atomic writes) and how to verify the guarantee holds. It assumes you have read the orchestration overview in orchestrating spatial pipelines with Airflow.

Prerequisites #

Before applying these patterns you need:

• Airflow 2.4+ with the TaskFlow API (@dag / @task), so tasks are plain Python functions and dependencies wire from calls.
• Python packages: geopandas, shapely, pyproj, and pyarrow for GeoParquet round-trips, plus your object-store client (boto3 or s3fs).
• An object store that supports atomic single-object writes and cheap existence checks — S3, GCS, or any store where “write one object” and “does this key exist” are both O(1). The GeoParquet data-lake layout this refresh writes into is the reference target.
• A stable version manifest. The routing graph_version and the demographic vintage must be resolvable at runtime, because they belong in the output key.

Configuration and execution parameters #

Idempotency depends on a handful of choices being made deliberately. The table lists them and the value that keeps re-runs safe.

The three techniques #

1. Deterministic output keys #

A task’s output must be addressed by what it is, not when it ran. Compute a key from the canonicalized inputs plus every code version that changes the result without changing the inputs. For a geospatial refresh those hidden inputs are the routing graph and the demographic vintage — omit them and a graph re-extract silently serves stale catchments under a key that looks current.

The key is a hash so it is fixed-length and path-safe. Two runs with identical inputs and versions collide on the same key by design; that collision is the cache hit that makes re-runs free.

2. Skip-if-exists #

Before computing, check whether the keyed output already exists in the store. If it does, the work is done — return the URI and skip. This is what converts a re-run from “redo everything” into “resume from the first gap”. It only works because the key is deterministic: a timestamped key would never match a prior run.

3. Atomic writes #

A task that writes directly to its final path can be killed mid-write, leaving a truncated GeoParquet that a downstream reader treats as real. Write to a temporary key first, then rename (an atomic metadata operation on object stores) once the write completes. For a database, wrap the load in a single transaction so a crash rolls back cleanly. Either way, a consumer only ever sees a complete artifact or nothing.

Annotated implementation #

The task below composes all three techniques. It generates drive-time catchments for one region, but the pattern is identical for any stage: derive a key, skip if the output exists, compute, and write atomically. CRS is asserted before any geometric operation and never assumed.

import hashlib
import geopandas as gpd
from pyproj import CRS
from airflow.decorators import task

STORAGE_CRS = CRS.from_epsg(4326)      # WGS84 interchange CRS
EQUAL_AREA_CRS = CRS.from_epsg(5070)   # Albers, for area verification only
GRAPH_VERSION = "osm-2026-06"


def output_key(region: str, input_uri: str, minutes: int,
               graph_version: str) -> str:
    """Content-addressed identity: same inputs + version -> same key."""
    fingerprint = f"{region}:{input_uri}:{minutes}:{graph_version}"
    return hashlib.sha256(fingerprint.encode()).hexdigest()[:16]


def atomic_write_parquet(gdf: gpd.GeoDataFrame, final_uri: str) -> None:
    """Write to a temp key, then rename. Readers never see a partial file."""
    import s3fs
    fs = s3fs.S3FileSystem()
    tmp_uri = final_uri + ".tmp"
    gdf.to_parquet(tmp_uri)            # full object materializes at tmp key
    fs.mv(tmp_uri, final_uri)          # atomic rename on the object store


@task(retries=3, retry_exponential_backoff=True)
def build_catchments(region: str, input_uri: str, minutes: int = 15) -> str:
    key = output_key(region, input_uri, minutes, GRAPH_VERSION)
    final_uri = f"s3://li-refresh/catchments/{key}.parquet"

    import s3fs
    fs = s3fs.S3FileSystem()
    if fs.exists(final_uri):           # skip-if-exists: the work is already done
        return final_uri

    gdf = gpd.read_parquet(input_uri)
    if gdf.crs is None:                # a dropped CRS must fail loudly
        raise ValueError(f"{input_uri}: missing CRS; refusing to route")
    gdf = gdf.to_crs(STORAGE_CRS)

    gdf["catchment"] = gdf.geometry.apply(
        lambda pt: solve_isochrone(pt.x, pt.y, minutes=minutes,
                                   graph_version=GRAPH_VERSION)
    )
    out = gdf.set_geometry("catchment")
    assert out.crs == STORAGE_CRS, "CRS drifted before write"
    atomic_write_parquet(out, final_uri)
    return final_uri

For a task whose sink is a database rather than object storage, the atomic-write technique becomes an upsert inside one transaction. The ON CONFLICT clause makes a second run overwrite the first row instead of appending a duplicate — the row’s natural key (site id plus run key) is what the conflict resolves on.

@task
def upsert_scored(scored_uri: str, run_key: str) -> None:
    import psycopg2, psycopg2.extras
    gdf = gpd.read_parquet(scored_uri)
    assert gdf.crs is not None, "scored frame lost its CRS"
    gdf = gdf.to_crs(STORAGE_CRS)
    rows = [(r.site_id, run_key, float(r.score), r.geometry.wkb_hex)
            for r in gdf.itertuples()]
    with psycopg2.connect(DSN) as conn, conn.cursor() as cur:   # one transaction
        psycopg2.extras.execute_values(cur, """
            INSERT INTO scored_sites (site_id, run_key, score, geom)
            VALUES %s
            ON CONFLICT (site_id, run_key)
            DO UPDATE SET score = EXCLUDED.score, geom = EXCLUDED.geom
        """, rows, template="(%s, %s, %s, ST_GeomFromWKB(decode(%s,'hex'), 4326))")
    # context-manager commit is atomic: a crash mid-load rolls back cleanly

The caching that makes overlapping trade areas cheap across weekly runs uses the same deterministic-key idea one layer down, at the routing engine; those caching strategies for repeated network queries and the skip-if-exists check here reinforce each other — the cache short-circuits the solve, and skip-if-exists short-circuits the whole task.

Failure modes and debugging #

The subtle failure is the timestamp in the key. It looks harmless — even reasonable — but it defeats the entire mechanism: every run produces a unique key, so skip-if-exists never fires, backfills recompute history, and the pipeline is idempotent in name only. Key strictly on the logical date and the pinned versions, never on wall-clock time.

Verification #

The definitive test of idempotency is that a second run produces identical output. Verify it with a content hash of the artifact, not a visual check.

1. Re-run yields an identical hash. Run the task twice against the same inputs and compare a stable hash of the output. Sort the frame first so row order cannot perturb the digest.
2. Row count is stable. A re-run must not change the row count of the DB sink — an increase means a non-idempotent append slipped through.
3. Geometry validity and CRS survive the round-trip. assert gdf.geometry.is_valid.all() and assert gdf.crs is not None after reading the written artifact back.
4. Area sanity in an equal-area CRS. Compare total catchment area between runs; it should be identical to floating-point tolerance.

import hashlib
import geopandas as gpd

def artifact_hash(uri: str) -> str:
    gdf = gpd.read_parquet(uri)
    assert gdf.crs is not None, "artifact lost its CRS"
    ordered = gdf.sort_values("site_id").reset_index(drop=True)
    # WKB encodes geometry deterministically; hash it alongside the attributes.
    payload = ordered.drop(columns="geometry").to_json().encode()
    payload += b"".join(g.wkb for g in ordered.geometry)
    return hashlib.sha256(payload).hexdigest()

first = artifact_hash(build_catchments("midwest", INPUT_URI))
second = artifact_hash(build_catchments("midwest", INPUT_URI))   # should skip
assert first == second, "task is not idempotent: outputs differ across runs"

An identical hash across two runs is the proof: the second run either skipped via the existence check or recomputed to the same bytes, and in both cases the pipeline is safe to re-run. That guarantee is what lets a backfill replay a year of intervals — or a mid-batch crash resume — without a human auditing what did and did not complete.

Related #

• Orchestrating Spatial Pipelines with Airflow — the full refresh DAG these idempotent tasks compose into.
• Configuring AWS S3 for Geospatial Data Lakes — the object-store layout the atomic writes target.
• Caching Strategies for Repeated Network Queries — deterministic keys applied one layer down at the routing engine.

← Back to Orchestrating Spatial Pipelines with Airflow