This guide is part of Orchestrating Spatial ETL Pipelines, the reference for turning spatial extract, clean, and load scripts into scheduled, restartable, observable data products.

Building Airflow DAGs for Spatial ETL

Apache Airflow is the default scheduler for spatial pipelines that run on a predictable cadence — nightly satellite passes, hourly sensor feeds, weekly portal refreshes — and lean on a broad operator ecosystem for S3, Postgres, and HTTP. But a spatial workload breaks the assumptions baked into most Airflow tutorials. The unit of work is a tile or a scene, not a warehouse row; the external dependencies are rate-limited catalog APIs, not an internal database; and a single failed download must not sink a 500-scene run. Naively translating a working notebook into one fat PythonOperator produces a DAG that double-loads tiles on retry, trips HTTP 429s across a fan-out, and takes a full-history re-run to reprocess yesterday.

The fix is to shape the DAG around the five-stage spatial ETL model and let Airflow’s structural primitives — task boundaries, pools, sensors, and dynamic task mapping — carry the concerns that the notebook papered over. The diagram below shows the canonical DAG graph: a sensor gates the run, Discover emits a manifest, Extract fans out under a bounded pool with retries, and the downstream stages hand off through object storage.

Each box is a separately retryable, separately observable task with an explicit contract for what it reads and writes. The arrows are not in-memory function calls — they are durable handoffs through S3, so any single stage can be reprocessed in isolation during a backfill.

Prerequisites & Environment

This guide targets Airflow 2.7 through 3.0 with the Amazon and Postgres provider packages, running Python 3.11. The provider packages ship the S3Hook and PostgresHook that give tasks credentialed access to object storage and the analytical database without hard-coding secrets.

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

Verify the environment resolves the providers and that the two connections the DAG relies on exist:

airflow version
airflow providers list | grep -E "amazon|postgres"
airflow connections get aws_default
airflow connections get postgis_analytics

Register the connections once, either through the UI or the CLI, so hooks resolve them by ID at runtime:

airflow connections add postgis_analytics \
  --conn-type postgres \
  --conn-host db.internal --conn-schema gis \
  --conn-login etl --conn-password "$PG_PASSWORD" --conn-port 5432

Version / Compatibility Matrix

Airflow’s task-authoring surface has shifted meaningfully across recent releases. The table maps each version to the features this guide uses and the caveats that bite spatial pipelines specifically.

Pin to a single minor version per environment. The largest jump is 2.x to 3.0: schedule_interval and execution_date are removed, so a DAG written against 2.x that references either will not import on 3.0 until you rename them to schedule and logical_date.

Step-by-Step Implementation

Step 1 — Gate the Run With a Sensor

A spatial DAG should not start Extract until the upstream data actually exists. A sensor blocks the run until new STAC items land in a time window or a file appears in a bucket. Use the reschedule mode (or a deferrable sensor on 2.7+) so a long poll frees the worker slot instead of holding it.

from __future__ import annotations

from datetime import datetime, timedelta

from airflow.decorators import task
from airflow.sensors.base import PokeReturnValue
from pystac_client import Client


@task.sensor(poke_interval=600, timeout=timedelta(hours=6), mode="reschedule")
def wait_for_new_stac_items(logical_date: datetime) -> PokeReturnValue:
    """Poll a STAC API until items exist for this run's day; return the manifest."""
    stac_client = Client.open("https://earth-search.aws.element84.com/v1")
    window = f"{logical_date:%Y-%m-%d}T00:00:00Z/{logical_date:%Y-%m-%d}T23:59:59Z"
    search = stac_client.search(collections=["sentinel-2-l2a"], datetime=window)
    item_ids = [item.id for item in search.items()]
    # is_done gates the DAG; the manifest is handed to the mapping step via XCom.
    return PokeReturnValue(is_done=bool(item_ids), xcom_value=item_ids)

This decorated sensor is the entry point for the STAC catalog sync workflow: the sensor decides whether to run, and the sync logic decides what to pull once it does.

Step 2 — Discover the Units of Work

Discover resolves the manifest — which tiles or scenes this run must process — and returns a list of small, JSON-serializable descriptors. It never returns the data itself, only keys, so the graph can be planned before a byte is downloaded.

@task
def discover_tiles(item_ids: list[str], logical_date: datetime) -> list[dict]:
    """Turn a list of scene IDs into per-tile work descriptors for mapping."""
    run_key = f"{logical_date:%Y%m%d}"
    return [
        {"scene_id": scene_id, "tile_id": scene_id.split("_")[1], "run_key": run_key}
        for scene_id in item_ids
    ]

Step 3 — Extract With Dynamic Task Mapping, a Pool, and Retries

This is the only stage that should carry aggressive retries, because its failures are transient — a dropped connection, an HTTP 429, a slow signed-URL redirect. Retrying Transform or Load instead re-runs deterministic logic against known-bad data. Dynamic task mapping (expand) creates one mapped task instance per manifest entry at runtime, and a pool caps how many run at once against the rate-limited endpoint.

import geopandas as gpd
from airflow.decorators import task
from airflow.providers.amazon.aws.hooks.s3 import S3Hook


@task(
    retries=4,
    retry_delay=timedelta(seconds=30),
    retry_exponential_backoff=True,
    max_retry_delay=timedelta(minutes=10),
    pool="stac_api",  # 4 slots — bounds concurrency against the rate-limited API
)
def extract_tile(work: dict) -> str:
    """Download one scene, write raw features to S3, and return the object key."""
    gdf: gpd.GeoDataFrame = _download_scene_features(work["scene_id"])
    key = f"raw/{work['run_key']}/{work['tile_id']}.parquet"
    hook = S3Hook(aws_conn_id="aws_default")
    with hook.get_conn() as _:  # ensures the connection resolves before writing
        gdf.to_parquet(f"s3://etl-spatial/{key}")
    return key

Only the Extract task overrides retries; the DAG default (set in Step 6) is retries=0. Because to_parquet writes a new object keyed on run_key and tile_id, a retry overwrites rather than appends — the serialization contract for handing spatial data between tasks is covered in passing GeoDataFrames between Airflow tasks.

Step 4 — Transform and Validate

Transform reads the raw keys back from S3, normalizes CRS, and repairs geometry using the Shapely 2.0 array API. Validate splits the batch into passing rows and a dead-letter set instead of raising, so one malformed scene never blocks the rest.

import shapely


@task
def transform_and_validate(key: str) -> dict:
    """Read raw features, repair geometry, split valid from dead-letter, persist both."""
    hook = S3Hook(aws_conn_id="aws_default")
    gdf = gpd.read_parquet(f"s3://etl-spatial/{key}").to_crs(4326)

    repaired = shapely.make_valid(gdf.geometry.values)  # vectorized, never .apply()
    gdf.geometry = repaired
    keep = shapely.is_valid(repaired) & ~gdf.geometry.isna().values

    clean_key = key.replace("raw/", "clean/")
    dead_key = key.replace("raw/", "dead/")
    gdf.loc[keep].to_parquet(f"s3://etl-spatial/{clean_key}")
    if (~keep).any():
        gdf.loc[~keep].to_parquet(f"s3://etl-spatial/{dead_key}")
    return {"clean_key": clean_key, "passed": int(keep.sum()), "dead": int((~keep).sum())}

Step 5 — Load With an Idempotent Hook Write

Load reads the validated keys and writes to PostGIS using the PostgresHook to resolve credentials. The write must be idempotent so a retried Load never duplicates geometries — the delete-by-key-then-insert transaction pattern is detailed in writing idempotent spatial ETL tasks in Airflow.

from airflow.providers.postgres.hooks.postgres import PostgresHook
from sqlalchemy import text


@task
def load_to_postgis(result: dict, table: str = "scenes") -> int:
    """Upsert one tile's clean features into PostGIS in a single transaction."""
    gdf = gpd.read_parquet(f"s3://etl-spatial/{result['clean_key']}")
    engine = PostgresHook(postgres_conn_id="postgis_analytics").get_sqlalchemy_engine()
    tile_ids = gdf["tile_id"].unique().tolist()
    with engine.begin() as conn:  # delete + insert atomic — safe under retry
        conn.execute(text(f"DELETE FROM {table} WHERE tile_id = ANY(:ids)"), {"ids": tile_ids})
        gdf.to_postgis(table, conn, if_exists="append", index=False)
    return len(gdf)

Step 6 — Assemble the DAG

The TaskFlow API wires the stages by passing return values, which Airflow records as XCom references. The DAG-level defaults set retries=0 so only Extract retries, and catchup=False because the pipeline tracks its own incremental window.

from airflow.decorators import dag


@dag(
    schedule="@daily",           # 3.0: 'schedule'; 2.x also accepts 'schedule_interval'
    start_date=datetime(2026, 1, 1),
    catchup=False,               # do not launch one run per historical interval
    default_args={"retries": 0, "owner": "geo-platform"},
    max_active_runs=1,           # serialize runs so watermark logic stays correct
    tags=["spatial", "etl"],
)
def spatial_etl():
    item_ids = wait_for_new_stac_items()
    work = discover_tiles(item_ids)
    raw_keys = extract_tile.expand(work=work)          # dynamic fan-out over tiles
    results = transform_and_validate.expand(key=raw_keys)
    load_to_postgis.expand(result=results)


spatial_etl()

Advanced Patterns & Edge Cases

Making Every Task Safe to Retry

Dynamic task mapping means a mapped Extract instance can fail and re-run while its siblings succeed, and Load can be retried after a database blip. Both must produce the same table state on the second run as the first. The delete-by-key-then-insert transaction and the atomic .part rename for file outputs — both keyed on logical_date — are the two building blocks, worked through in writing idempotent spatial ETL tasks in Airflow.

Never Push GeoDataFrames Through XCom

The TaskFlow wiring above returns S3 keys, not GeoDataFrames, on purpose. XCom serializes return values into the metadata database, and a GeoDataFrame of even a few thousand geometries overruns the default XCom size limit and bloats the database. The correct pattern — serialize to GeoParquet in object storage and pass only the URI — is detailed in passing GeoDataFrames between Airflow tasks.

Sizing Pools and Catchup for Incremental Windows

A pool bounds concurrency against a rate-limited API, but the deeper scheduling question is which windows have already run. With catchup=False the DAG only processes the current interval, which means historical gaps need an explicit backfill strategy driven by a watermark — the mechanics of tracking processed windows and issuing bounded catch-up runs are covered in scheduling and incremental spatial loads.

Performance Optimization

The single biggest throughput lever in a mapped spatial DAG is matching pool size to the endpoint’s real concurrency budget, not to worker count. Oversizing the pool trips HTTP 429s that cascade retries and reduce effective throughput; undersizing wastes workers. The table below shows a 500-tile Sentinel-2 extract measured against a public STAC endpoint with a soft limit of roughly 5 concurrent requests.

# Size the pool to the measured knee, not the worker count.
# airflow pools set stac_api 4 "bounded concurrency for rate-limited STAC API"
from airflow.decorators import task


@task(pool="stac_api", pool_slots=1, max_active_tis_per_dag=4)
def extract_tile_tuned(work: dict) -> str:
    """Same extract, with per-DAG mapped-instance concurrency capped to the knee."""
    return _extract_and_store(work)

The knee sits at 4 slots: concurrency past that point spends its time retrying throttled requests rather than downloading. Set the pool once and every mapped Extract instance across every run respects it.

Integration into ETL Pipelines

Wire the DAG’s S3 layout to the same partitioning the ingestion and cleaning references produce: key raw and clean objects on run_key plus tile_id so a backfill of one day touches only that day’s objects. Route the dead-letter Parquet from Step 4 to a dedicated prefix and add a lightweight downstream task that alerts on dead-letter volume rather than on individual records, so one corrupt scene raises a metric instead of failing the run. Emit structured logs at every task boundary with the tile_id attached — Airflow’s task logs are the audit trail that answers “did tile 31UFT load last night?” without reading a stack trace. For event-driven variants, replace the time-based schedule with a Dataset (Airflow 2.8+) or asset (3.0) that fires the DAG when an upstream producer publishes new scenes, and let the same incremental scheduling and watermark logic decide which windows are still outstanding.

Troubleshooting Reference

Related

• Writing Idempotent Spatial ETL Tasks in Airflow — delete-by-key-then-insert transactions and atomic .part renames keyed on logical_date
• Passing GeoDataFrames Between Airflow Tasks — why GeoDataFrames must not travel through XCom and the GeoParquet-URI handoff pattern
• Orchestrating Spatial ETL Pipelines — the reference model and framework comparison this Airflow guide sits within
• Syncing STAC Catalogs with pystac-client — the catalog-sync logic the sensor and Extract stages orchestrate
• Scheduling and Incremental Spatial Loads — watermark tracking and catch-up runs that complement catchup=False