Location Intelligence Retail Site Selection Automation

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.

Choice Setting Why it matters
Output identity hash of inputs + graph_version + vintage Same inputs → same key → skip on re-run
Write mode (files) temp key + atomic rename No consumer ever sees a half-written object
Write mode (DB) upsert (ON CONFLICT DO UPDATE) Re-run converges instead of duplicating rows
Existence check head_object before compute Turns a retry into a resume
Hash length 16 hex chars (64-bit) Collision-safe at pipeline scale, short in paths
CRS on read assert then to_crs A dropped CRS must fail loudly, not compute wrong
Retry scope retries=3, exponential backoff Only transient failures retry; permanent ones dead-letter

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.

The same task run twice, with and without a deterministic key A task that appends to a timestamped file writes 4,610 rows, fails after the write, and on retry writes 4,610 more into a second file — leaving 9,220 rows that the next stage happily sums. The same task writing to a key derived from the run date overwrites its own partial output, so the retry leaves exactly 4,610. The retry is not the bug — the key is key includes a timestamp attempt 1 → trade_areas_1421.parquet 4,610 rows written, then the task dies attempt 2 → trade_areas_1438.parquet 4,610 rows written, task succeeds the next stage reads 9,220 rows every store counted twice, no error anywhere key derived from the run attempt 1 → date=2026-03-14/part-0.parquet partial write, task dies, nothing promoted attempt 2 → same path, written to a temp key then moved into place atomically the next stage reads 4,610 rows the retry is invisible downstream Doubled rows are the friendly version of this failure — they are at least visible in a count. The quiet version is a partially-written file that parses, and a market whose demographics are half a month old.

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.

python
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.

python
@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.

Write aside, verify, then promote The task writes to a temporary key, runs its validation against what it just wrote, and only then updates the manifest entry that readers follow. A reader arriving at any moment sees either the previous complete version or the new complete version, never a half-written one. Readers follow a pointer, so the swap is the only moment that matters 1 · write aside _tmp/run=2026-03-14/ 2 · validate counts · CRS · geometry 3 · promote manifest points at the new key What a reader sees during each step steps 1 and 2: the previous version, complete · step 3: the new version, complete · never anything in between Failure at step 1 or 2 leaves an orphaned temporary key, which a scheduled cleanup removes. That is the whole cost of the pattern, and it buys the guarantee that no consumer ever reads a partial refresh. Where the store offers a conditional write, use it for the promote step so two concurrent runs cannot both win.

Failure modes and debugging

Symptom Cause Fix
Duplicate rows after a retry INSERT-only load, no conflict clause Upsert on a natural key inside one transaction.
Truncated / unreadable GeoParquet Task killed while writing the final path Write to a .tmp key, then atomic rename.
Re-run recomputes everything Key includes a timestamp or datetime.now() Key on the logical date and pinned versions only.
Stale catchments under a fresh-looking key graph_version omitted from the key Add every output-changing version to the fingerprint.
Planar-nonsense area or empty join CRS dropped on a GeoParquet round-trip Assert crs on read; to_crs before any geometry op.
Skip fires but output is wrong Key collides across genuinely different inputs Widen the fingerprint to include the distinguishing input.

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.

Backfill as a fan-out over logical dates Six historical months are refreshed in parallel, each writing to its own dated partition, while the daily run for today proceeds untouched. Because every task's output key comes from its logical date rather than from the wall clock, the backfill and the daily run cannot collide. Idempotency is what makes a backfill boring backfill, six logical dates in parallel 2025-10 2025-11 2025-12 2026-01 2026-02 2026-03 today's scheduled run, unaffected 2026-03-14 writes its own dated partition; shares no mutable state with the six above The one thing that breaks this is a task that reads "the current version" of an upstream dataset instead of the version that was current on its logical date — then a backfill silently rewrites history with today's inputs. Pin upstream versions per logical date and the backfill reproduces the past instead of overwriting it.

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.
python
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.

Frequently Asked Questions

Is a delete-then-write task idempotent?

Only if the delete and the write are a single atomic step, which on an object store they are not. The window between them is a period where the dataset simply does not exist, and any reader arriving in it gets an empty result rather than an error — which is worse, because empty results propagate quietly through aggregations. Writing to a new key and swapping a pointer avoids the window entirely, and it leaves the previous version available if the new one turns out to be wrong.

How does a task know it has already run?

By checking for its own output at the deterministic key it would write, not by consulting a status flag. A flag can be set by a task that then dies before completing the write, and it can be cleared by a hand that meant to clear something else. The output’s existence, ideally with a completion marker written last, is the only evidence that survives every failure mode — and it makes a rerun after a partial failure both safe and cheap.

What about tasks that call an external service?

Make the call idempotent from your side, since you cannot make the service so. Derive a request key from the inputs, record the response under it, and check the record before calling again — that turns a repeated routing request into a cache hit rather than a second charge and a slightly different polygon. For services that genuinely mutate remote state, ask for an idempotency key if the API offers one; where none exists, isolate the call in its own task so a retry of the surrounding work does not repeat it.

← Back to Orchestrating Spatial Pipelines with Airflow