This guide is part of Building Airflow DAGs for Spatial ETL, a chapter of the broader Orchestrating Spatial ETL Pipelines reference.

Writing Idempotent Spatial ETL Tasks in Airflow

The Problem: Retries Turn Append-Style Loads Into Duplicate Geometries

An Airflow task will eventually run more than once — that is what retries and manual clears are for. A spatial load task is only safe under that guarantee if its second run produces exactly the same table state as its first. A task written as a plain append violates this, and the failure is silent until a downstream spatial join returns doubled counts.

• Commit-then-crash duplicates rows. If the insert commits but the worker is killed before Airflow records success, the retry appends the batch’s geometries a second time — the database has no memory that this tile_id was already loaded.
• Partial writes masquerade as complete. A GeoParquet write killed halfway leaves a truncated object at the final key. The next run sees the key exists, skips the tile, and a gap enters the mosaic that no error ever reported.
• Mapped fan-out multiplies the blast radius. With dynamic task mapping, one mapped instance can retry while its siblings succeed, so non-idempotent logic corrupts a single tile’s rows inside an otherwise-green run — the hardest kind of bug to spot.
• Backfills replay history. Re-running last month to fix a Transform bug re-executes every Load. If Load appends, the backfill doubles a month of data instead of refreshing it.

Version and Environment Compatibility

pip install "apache-airflow==2.9.3" "geopandas>=1.0" "pyarrow>=15.0" \
  "sqlalchemy>=2.0" "apache-airflow-providers-amazon>=8.0"

The recipe below assumes the delete and insert run against PostGIS through the PostgresHook and that GeoParquet outputs land in S3 through the S3Hook, exactly as wired in the parent DAG.

Recipe: load_scene_idempotent — Transactional Upsert Plus Atomic Rename

The single function below makes a spatial load safe to retry on two fronts at once: it replaces the batch’s rows in PostGIS inside one transaction, and it persists the same features to GeoParquet through a .part object that is renamed only after the write completes. Both operations key on logical_date combined with tile_id, so every attempt of a given run targets identical rows and objects.

from __future__ import annotations

import logging
from datetime import datetime

import geopandas as gpd
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.postgres.hooks.postgres import PostgresHook
from sqlalchemy import text

logger = logging.getLogger(__name__)


def load_scene_idempotent(
    gdf: gpd.GeoDataFrame,
    logical_date: datetime,
    *,
    table: str = "scenes",
    key_col: str = "tile_id",
    bucket: str = "etl-spatial",
    postgres_conn_id: str = "postgis_analytics",
    aws_conn_id: str = "aws_default",
) -> dict[str, int | str]:
    """Load a batch into PostGIS and GeoParquet so retries never double-load.

    The database write is a delete-by-key-then-insert in one transaction; the
    file write goes to a `.part` object renamed on success. Both are keyed on
    ``logical_date`` + ``key_col`` so every retry of this run is a no-op replace.
    """
    if gdf.empty:
        logger.info("Empty batch for %s; nothing to load.", logical_date.date())
        return {"rows": 0, "object_key": ""}

    run_key = logical_date.strftime("%Y%m%d")
    batch_keys = gdf[key_col].unique().tolist()

    # --- Database: atomic delete-then-insert (idempotent upsert) ---------------
    engine = PostgresHook(postgres_conn_id=postgres_conn_id).get_sqlalchemy_engine()
    with engine.begin() as conn:  # begin() commits on exit, rolls back on error
        deleted = conn.execute(
            text(f"DELETE FROM {table} WHERE {key_col} = ANY(:ids)"),
            {"ids": batch_keys},
        ).rowcount
        gdf.to_postgis(table, conn, if_exists="append", index=False)
    logger.info(
        "Upserted %d rows into %s (replaced %d) for run %s",
        len(gdf), table, deleted, run_key,
    )

    # --- File: write to .part, then atomic rename on the object store ----------
    hook = S3Hook(aws_conn_id=aws_conn_id)
    final_key = f"clean/{run_key}/{'-'.join(map(str, batch_keys))}.parquet"
    part_key = f"{final_key}.part"
    try:
        gdf.to_parquet(f"s3://{bucket}/{part_key}")  # transient object
        # copy+delete is S3's atomic rename; the final key appears whole or not at all
        s3 = hook.get_conn()
        s3.copy_object(
            Bucket=bucket,
            CopySource={"Bucket": bucket, "Key": part_key},
            Key=final_key,
        )
        s3.delete_object(Bucket=bucket, Key=part_key)
    except Exception:
        # leave no partial final object behind on failure
        hook.delete_objects(bucket=bucket, keys=[part_key])
        logger.exception("GeoParquet write failed for run %s; .part cleaned up", run_key)
        raise

    return {"rows": len(gdf), "object_key": final_key}

Key Implementation Notes

• engine.begin() is the atomicity boundary. It opens a transaction that commits when the with block exits cleanly and rolls back on any exception. Because DELETE and to_postgis share that block, a crash between them cannot leave the table with the old rows deleted and the new ones missing — the database is only ever in the before-state or the after-state.
• ANY(:ids) deletes exactly the batch’s keys, nothing more. Scoping the delete to batch_keys means a mapped task instance handling tile 31UFT never touches tile 32ULC’s rows, so sibling mapped instances stay independent even when they write to the same table.
• logical_date is the stable per-run seed, not datetime.now(). Every retry and manual clear of a run shares one logical_date, so run_key is identical across attempts. Seeding the object key from wall-clock time would mint a new key on each retry and defeat the whole scheme.
• S3 has no in-place rename, so copy-then-delete is the atomic swap. The final key is created by a single copy_object from the .part; readers see it appear whole. On a POSIX filesystem the equivalent is Path.rename, which is atomic within a filesystem — use whichever the target store provides.
• The .part is cleaned up on failure. The except branch deletes the temporary object so a failed attempt leaves neither a partial final file nor orphaned .part litter for the retry to trip over.
• Return values are small and JSON-serializable. The function returns a row count and the object key — never the GeoDataFrame — so it composes with TaskFlow XCom without breaching size limits.

Troubleshooting Common Idempotency Failures

Integration Note

load_scene_idempotent is the concrete implementation of Step 5 in building Airflow DAGs for spatial ETL: drop it into the @task that follows Transform, pass the DAG’s logical_date, and set the DAG default to retries=0 so only Extract retries while Load stays safe under a manual clear. Because the function takes a GeoDataFrame in memory but returns only an object key, it pairs directly with the serialization boundary described in passing GeoDataFrames between Airflow tasks — the upstream task hands this one a URI, this one reads the GeoParquet, upserts, and hands the next task another URI.

Related

• Building Airflow DAGs for Spatial ETL — the parent DAG where this idempotent load slots in as the Load stage
• Passing GeoDataFrames Between Airflow Tasks — the GeoParquet-URI handoff that feeds this task its input
• Orchestrating Spatial ETL Pipelines — the idempotency and exactly-once-load principles this recipe implements