Scheduling a Daily Satellite Ingest DAG in Airflow

Build a daily Airflow DAG for satellite ingest: deferrable sensors, dynamic task mapping over tiles, staged writes, a validation gate, backfills and pool-based rate limiting.

A daily ingest DAG has five parts: a deferrable sensor that waits for the scene to be complete, a task that builds the tile list, a mapped task per tile that writes to a staging prefix, a validation gate, and a publish step that promotes the mosaic atomically. Everything else — retries, pools, backfills — is configuration around that spine.

This page is the concrete DAG. For the design reasoning behind each part, see Orchestrating Geospatial ML Pipelines.

Five tasks, one fan-out, one gate The DAG begins with a deferrable sensor waiting for the source scene. It feeds a task that builds the tile list, which expands into many mapped per-tile tasks. Those converge on a validation task that checks coverage and statistics, and only if validation passes does the publish task promote the staged mosaic. wait → plan → fan out → gate → publish sensor deferrable tile list 6 000 ids process_tile[0] process_tile[1] process_tile[n] validate all_done trigger publish atomic promote Only the last task changes anything a consumer can see.

Why This Fails in Geospatial ML Pipelines

The scheduler outage is the most embarrassing failure and the easiest to cause. Airflow parses every DAG file on a short interval, in the scheduler process, which has only Airflow’s dependencies. A DAG file with import rasterio at the top therefore fails to import, and a DAG that fails to import is not scheduled — it does not alert, it does not run, it simply disappears from the list. Weeks of missing predictions have been traced to exactly this.

The second is sensors eating the cluster. A PythonSensor in the default poke mode occupies a worker slot for its entire wait. On a DAG that waits four hours for a scene, sixteen worker slots means four concurrent runs at most, and a backfill deadlocks: every slot is held by a sensor waiting for data that the tasks behind those sensors would have produced.

Third, backfills that outrun the source. catchup=True with default concurrency launches one run per missed day simultaneously. Thirty runs, each fanning out over six thousand tiles, will exhaust any imagery provider’s rate limit, and the resulting 429 responses look like transient network errors — so the retries make it worse.

Core Principles

  • Keep heavy imports inside task functions. The scheduler must never need GDAL.
  • Use deferrable or reschedule-mode sensors. Never hold a slot to wait.
  • Resolve the model version once, in the first task, and pass it down.
  • Bound concurrency where the constraint is — API pool, not global workers.
  • Gate publication on validation with trigger_rule="all_done" so partial failures are inspected, not published.
  • Pass keys, not arrays, through XComs.

Production-Ready Code

"""dags/landcover_daily.py — parsed by the scheduler, so keep imports light."""
from __future__ import annotations

from dataclasses import asdict, dataclass
from datetime import datetime, timedelta

from airflow.decorators import dag, task
from airflow.exceptions import AirflowSkipException
from airflow.sensors.python import PythonSensor

DEFAULT_ARGS = {
    "owner": "geo-ml",
    "retries": 3,
    "retry_delay": timedelta(minutes=2),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
}


@dataclass(frozen=True)
class RunSpec:
    scene_date: str
    collection: str
    model_uri: str
    model_version: str
    grid_id: str
    run_id: str


def _scene_ready(**context) -> bool:
    """Sensor callable — imports live here, not at module level."""
    from pystac_client import Client                     # noqa: PLC0415

    date = context["logical_date"].date().isoformat()
    catalog = Client.open("https://earth-search.aws.element84.com/v1")
    items = list(catalog.search(collections=["sentinel-2-l2a"],
                                bbox=[5.8, 47.2, 15.1, 55.1],
                                datetime=f"{date}/{date}").items())
    return bool(items) and all(len(i.assets) >= 12 for i in items)


@dag(
    dag_id="landcover_daily",
    schedule="0 4 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=True,
    max_active_runs=2,
    default_args=DEFAULT_ARGS,
    tags=["geospatial", "inference"],
)
def landcover_daily():

    wait_for_scene = PythonSensor(
        task_id="wait_for_scene",
        python_callable=_scene_ready,
        mode="reschedule",                 # releases the worker slot between pokes
        poke_interval=600,
        timeout=60 * 60 * 10,
        soft_fail=True,                    # a genuinely missing day skips, not fails
    )

    @task
    def build_spec(**context) -> dict:
        from geoml.registry import resolve_model         # noqa: PLC0415
        uri, version = resolve_model("landcover", stage="production")
        return asdict(RunSpec(
            scene_date=context["logical_date"].date().isoformat(),
            collection="sentinel-2-l2a",
            model_uri=uri, model_version=version,
            grid_id="utm33n_10m", run_id=context["run_id"],
        ))

    @task
    def list_tiles(spec: dict) -> list[str]:
        from geoml.grids import tiles_for_grid           # noqa: PLC0415
        tiles = tiles_for_grid(spec["grid_id"])
        if not tiles:
            raise AirflowSkipException("no tiles for this grid")
        return tiles

    @task(pool="imagery_api", max_active_tis_per_dag=32)
    def process_tile(spec: dict, tile_id: str) -> str:
        from geoml.inference import run_tile             # noqa: PLC0415
        return run_tile(RunSpec(**spec), tile_id)        # -> staged object key

    @task(trigger_rule="all_done")
    def validate(spec: dict, keys: list[str]) -> dict:
        from geoml.validation import validate_staged     # noqa: PLC0415
        report = validate_staged(RunSpec(**spec), [k for k in keys if k])
        if not report["ok"]:
            raise ValueError(f"validation failed: {report['failures'][:3]}")
        return report

    @task
    def publish(spec: dict, report: dict) -> None:
        from geoml.publish import promote                # noqa: PLC0415
        promote(RunSpec(**spec), n_tiles=report["n_tiles"])

    spec = build_spec()
    wait_for_scene >> spec
    keys = process_tile.partial(spec=spec).expand(tile_id=list_tiles(spec))
    publish(spec, validate(spec, keys))


landcover_daily()

Step-by-Step Walkthrough

Step 1 — Prove the DAG imports without the geospatial stack

python -c "import ast,sys; ast.parse(open('dags/landcover_daily.py').read())"
airflow dags list | grep landcover_daily
airflow dags list-import-errors

Run these three in CI on every commit. The third is the one that catches the silent outage.

Step 2 — Create the pool that bounds the source API

airflow pools set imagery_api 24 "Concurrent requests to the imagery provider"

The pool is what makes a thirty-day backfill safe: however many runs are active, at most 24 tile tasks talk to the provider at once.

Step 3 — Test one task in isolation

airflow tasks test landcover_daily process_tile 2026-08-04

tasks test runs the callable outside the scheduler with a real context, which is the fastest way to debug a worker-side import or credential problem.

Step 4 — Run a bounded backfill

airflow dags backfill landcover_daily \
    --start-date 2026-07-01 --end-date 2026-07-31 \
    --max-active-runs 2 --reset-dagruns

With max_active_runs=2 and the pool in place, this walks the month at a rate the provider tolerates.

Step 5 — Alert on the gate, not on every task

from airflow.providers.slack.notifications.slack import send_slack_notification

validate_task = validate.override(
    on_failure_callback=[send_slack_notification(
        slack_conn_id="slack", channel="#geo-ml",
        text="landcover_daily validation failed for  — nothing published")],
)

A single tile failing and being retried is noise. Validation failing means the day’s mosaic will not publish, which is what a human needs to see.

A poke-mode sensor holds its worker slot the whole time Two timelines show sixteen worker slots over a four hour wait. In poke mode, four slots are permanently occupied by sensors and only twelve remain for real work. In reschedule mode the sensors occupy a slot only momentarily at each check, leaving nearly all slots available. Worker slot occupancy during a four-hour wait poke 4 slots held continuously 12 usable reschedule a slot only at each 10-minute poke ~16 usable With deferrable sensors the wait moves to the triggerer and costs no worker slot at all. In poke mode a backfill can deadlock on its own sensors.

Verification

import pytest
from airflow.models import DagBag


def test_dag_imports_without_geospatial_deps():
    """The scheduler has no rasterio — the DAG file must not need it."""
    bag = DagBag(dag_folder="dags/", include_examples=False)
    assert not bag.import_errors, bag.import_errors
    assert "landcover_daily" in bag.dags


def test_no_module_level_heavy_imports():
    src = open("dags/landcover_daily.py").read().split("def ")[0]
    for banned in ("import rasterio", "import geopandas", "import torch", "from osgeo"):
        assert banned not in src, f"{banned} at module level breaks the scheduler"


def test_sensor_does_not_hold_a_slot():
    dag = DagBag("dags/", include_examples=False).dags["landcover_daily"]
    sensor = dag.get_task("wait_for_scene")
    assert sensor.mode in ("reschedule", "deferrable")


def test_publish_depends_on_validation():
    dag = DagBag("dags/", include_examples=False).dags["landcover_daily"]
    upstream = {t.task_id for t in dag.get_task("publish").upstream_list}
    assert "validate" in upstream


def test_tile_task_is_pooled():
    dag = DagBag("dags/", include_examples=False).dags["landcover_daily"]
    assert dag.get_task("process_tile").pool == "imagery_api"

These five run in a second and cover every failure described above. The DagBag test in particular belongs in CI, because a DAG that stops importing produces no alert of its own.

Bound the backfill or the provider will bound it for you Two backfill strategies over a thirty day range are compared. Unbounded catchup launches all thirty runs at once and the source API returns rate limit errors. With max active runs capped at two and a pool on the download task, runs proceed steadily and complete without errors. 30-day backfill, two configurations unbounded … 30 runs at once → HTTP 429 everywhere capped 2 at a time, pool caps the API → completes The rate limit belongs to the provider, not to your cluster — express it as a pool. Retries against a rate limit make the problem worse, not better.

FAQ

Why does my sensor block every worker slot?

Because poke mode holds the slot for the whole wait. Use mode="reschedule", which releases it between checks, or a deferrable sensor, which moves the wait to the triggerer and uses no worker slot at all.

How should catchup and backfills be configured?

Keep catchup=True so gaps are filled deliberately, but cap max_active_runs at two or three and put the download task in a pool. Otherwise a month-long backfill launches thirty concurrent fan-outs at the provider.

Why does the scheduler fail to import my DAG?

Because the file imports the geospatial runtime at module level. The scheduler parses DAG files constantly with only Airflow’s dependencies — move rasterio, GDAL and model imports inside the task functions, as the code above does.

Part of: Orchestrating Geospatial ML Pipelines Part of: Geospatial MLOps and Model Deployment