Orchestrating Spatial Pipelines with Airflow

Turning a location-intelligence refresh from a pile of ad-hoc scripts into a scheduled, resumable, auditable system means expressing the whole ingest → score → export chain as a single Airflow DAG whose every task is safe to re-run.

A retail site-selection platform is not one computation; it is a dependency graph. Coordinates arrive, get validated, drive isochrone generation, join against demographics, feed a suitability model, and finally export to the formats stakeholders consume. Run any stage out of order — or re-run one after a mid-batch crash without idempotency — and you silently corrupt the numbers a capital committee will act on. This stage of the Location Intelligence Architecture & Data Foundations discipline is about making that graph deterministic: an orchestrator that resumes a half-finished run without recomputing finished work, retries only what is transient, and quarantines what is broken instead of poisoning the batch.

Concept: the refresh as a directed acyclic graph #

Airflow models work as a DAG — a set of tasks with directed dependency edges and no cycles. The scheduler walks that graph, running a task only once all of its upstream dependencies have succeeded, and it does so per logical date: each scheduled interval (a daily or weekly window) is a distinct run identified by a data interval, not by wall-clock time. That distinction is the foundation of everything reproducible here. A task’s behavior must depend only on its logical date and its inputs, never on “now”, because Airflow may run today’s interval tomorrow (a catch-up), or re-run last month’s interval during a backfill, and the output must be identical either way.

The location-intelligence refresh maps cleanly onto six stages. Ingestion pulls new and changed store coordinates; a validation gate rejects malformed geometry before it can propagate; isochrone generation turns each surviving coordinate into a drive-time catchment; a spatial join attributes demographics to each catchment; a scoring stage collapses those attributes into a ranked suitability figure; and an export stage writes the formats downstream consumers read. Each edge in that chain is a hard dependency — scoring cannot begin until the join has produced attributed catchments — and each node is a unit of retry, logging, and idempotency.

The property that makes this graph trustworthy is idempotency: re-executing a task with the same inputs yields the same output and the same side effects, so a retry or a backfill never double-writes or drifts. Formally, for a task $f$ and an input state $x$, idempotency requires $f(f(x)) = f(x)$. The whole pipeline is engineered so that partial failure plus retry is indistinguishable from a clean first run — the detailed techniques live in writing idempotent Airflow DAGs for geospatial refresh.

Architecture: the run key and the storage contract #

Every artifact this DAG produces is addressed by a deterministic run key rather than a timestamp. The key is a hash of the logical date, the input snapshot, and the versions of the two things that change output without changing coordinates: the routing graph_version and the demographic vintage (for example the ACS release year). Two runs with identical inputs and identical versions produce byte-identical keys, so the storage layer becomes a content-addressable cache: a task checks whether its output key already exists and skips the compute if it does. This is the mechanism that makes backfills cheap and retries safe.

The storage contract sits on the same object-store foundation as the rest of the platform. Intermediate artifacts land in partitioned GeoParquet under a region × logical_date layout so a downstream task or a backfill reads only the partitions it needs; the partitioning conventions are detailed in partitioning GeoParquet by region and date. Durable, queryable state — the scored catchments a BI tool hits live — lands in PostGIS, written transactionally so a mid-write crash never leaves a half-populated table.

Airflow tasks pass small handles between each other over XCom — object-store URIs and row counts, never the geometry itself. Serializing a multi-region GeoDataFrame through the metadata database is a reliable way to melt it; the geometry travels through object storage, and only the pointer travels through the scheduler.

Configuration parameters #

The table below lists the DAG- and task-level settings that materially change scheduling behavior, throughput, and failure semantics. Treat anything outside these bands as a documented exception, not a convenience.

Two settings deserve emphasis. catchup=False is the correct default for a freshness-driven refresh — you want the latest data, not thirty replays of missed weeks — but a genuine historical rebuild flips it to True and lets the deterministic run key make the replay idempotent. And pool is the single most effective guard against a self-inflicted denial of service: without it, a wide dynamic mapping fans out hundreds of simultaneous isochrone requests and knocks over the very engine the pipeline depends on.

Step-by-step: the TaskFlow DAG #

The DAG below uses Airflow’s TaskFlow API (@dag / @task), which wires dependencies from ordinary Python function calls and passes handles automatically. Each stage is written to be idempotent, retries are scoped to transient failures, and dynamic task mapping fans the heavy stages out over regions. Note the explicit CRS handling: the validation and join stages assert a coordinate reference system before any geometric predicate runs, exactly as the coordinate rules in data validation rules for store coordinates demand.

from datetime import timedelta
from airflow.decorators import dag, task
from airflow.exceptions import AirflowFailException
import pendulum

import geopandas as gpd
from pyproj import CRS

STORAGE_CRS = CRS.from_epsg(4326)      # WGS84 for interchange
EQUAL_AREA_CRS = CRS.from_epsg(5070)   # Albers, for any area math
GRAPH_VERSION = "osm-2026-06"          # bump forces clean recompute
ACS_VINTAGE = "acs5-2024"

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": timedelta(minutes=30),
    "execution_timeout": timedelta(hours=2),
}


@dag(
    schedule="0 4 * * 1",              # weekly, Monday 04:00 UTC
    start_date=pendulum.datetime(2026, 1, 5, tz="UTC"),
    catchup=False,
    max_active_runs=1,
    default_args=default_args,
    tags=["location-intelligence", "refresh"],
)
def site_suitability_refresh():

    @task
    def run_key(**ctx) -> str:
        """Deterministic identity for this run: logical date + code versions."""
        import hashlib
        logical = ctx["data_interval_start"].to_date_string()
        raw = f"{logical}:{GRAPH_VERSION}:{ACS_VINTAGE}"
        return hashlib.sha256(raw.encode()).hexdigest()[:16]

    @task
    def ingest(key: str) -> list[str]:
        """Pull changed coordinates; return per-region GeoParquet URIs."""
        regions = discover_regions()               # e.g. ["midwest", "south", ...]
        uris = []
        for region in regions:
            out = f"s3://li-refresh/ingest/{key}/{region}.parquet"
            if object_exists(out):                 # idempotent skip
                uris.append(out)
                continue
            gdf = pull_changed_coordinates(region, crs=STORAGE_CRS)
            gdf.to_parquet(out)                    # atomic single-object write
            uris.append(out)
        return uris

    @task
    def validate(uri: str, key: str) -> str:
        """Hard gate. Bad rows go to the dead-letter store, not downstream."""
        gdf = gpd.read_parquet(uri)
        if gdf.crs is None:
            raise AirflowFailException(f"{uri}: missing CRS")   # non-retryable
        gdf = gdf.to_crs(STORAGE_CRS)
        in_bounds = gdf.geometry.x.between(-180, 180) & gdf.geometry.y.between(-90, 90)
        bad = gdf[~(gdf.geometry.is_valid & in_bounds)]
        if not bad.empty:
            bad.to_parquet(f"s3://li-refresh/dead-letter/{key}/{region_of(uri)}.parquet")
        clean = gdf[gdf.geometry.is_valid & in_bounds]
        out = f"s3://li-refresh/valid/{key}/{region_of(uri)}.parquet"
        clean.to_parquet(out)
        return out

    @task(pool="routing_pool", max_active_tis_per_dag=16)
    def isochrones(uri: str, key: str) -> str:
        """Generate drive-time catchments; cache hits short-circuit the solve."""
        gdf = gpd.read_parquet(uri)
        assert gdf.crs == STORAGE_CRS, "CRS drifted before routing"
        gdf["catchment"] = gdf.geometry.apply(
            lambda pt: solve_isochrone(pt.x, pt.y, minutes=15,
                                       graph_version=GRAPH_VERSION)
        )
        out = f"s3://li-refresh/catchments/{key}/{region_of(uri)}.parquet"
        gdf.set_geometry("catchment").to_parquet(out)
        return out

    @task
    def join_demographics(uri: str, key: str) -> str:
        catch = gpd.read_parquet(uri)
        bg = load_block_groups(ACS_VINTAGE)                 # demographic layer
        assert catch.crs == bg.crs, "CRS mismatch before spatial join"
        joined = gpd.sjoin(catch, bg, predicate="intersects", how="left")
        out = f"s3://li-refresh/attributed/{key}/{region_of(uri)}.parquet"
        joined.to_parquet(out)
        return out

    @task
    def score(uris: list[str], key: str) -> str:
        frames = [gpd.read_parquet(u) for u in uris]
        national = gpd.GeoDataFrame(pd.concat(frames, ignore_index=True),
                                    crs=STORAGE_CRS)
        national["score"] = suitability_score(national)     # deterministic model
        out = f"s3://li-refresh/scored/{key}.parquet"
        national.to_parquet(out)
        return out

    @task
    def export(uri: str, key: str) -> None:
        gdf = gpd.read_parquet(uri)
        publish_geopackage(gdf, key)     # atomic: write temp, then rename
        upsert_postgis(gdf, key)         # single transaction, ON CONFLICT upsert

    key = run_key()
    ingested = ingest(key)
    validated = validate.partial(key=key).expand(uri=ingested)
    routed = isochrones.partial(key=key).expand(uri=validated)
    attributed = join_demographics.partial(key=key).expand(uri=routed)
    scored = score(uris=attributed, key=key)
    export(uri=scored, key=key)


site_suitability_refresh()

The .expand() calls are dynamic task mapping: Airflow creates one task instance per region at runtime, so adding a market to discover_regions() widens the fan-out without touching the DAG structure. Each mapped instance is independently retryable — a routing failure in the South region does not force the Midwest to recompute. The partial(key=key) binds the shared run key across every mapped instance so they all address the same content-addressed run.

The run_key task is the linchpin. Because it hashes the logical date and the code versions rather than reading the clock, a backfill of an old interval reconstructs exactly the key that interval produced originally, and every downstream object_exists check short-circuits. That is what lets a failed run resume from the first uncomputed region instead of from zero.

Edge cases and failure modes #

Orchestration failures are rarely loud crashes; they are the quiet corruptions that survive into a report.

• Partial failure mid-fan-out. If the routing engine dies after three of eight regions finish, a naive re-run redoes all eight. The object_exists guard in each task turns the retry into a resume: finished regions are skipped, only the failed ones recompute. This depends entirely on the deterministic run key being stable across attempts.
• Non-idempotent writes. An INSERT-only load duplicates rows on every retry. Every terminal write here is an upsert (ON CONFLICT ... DO UPDATE) or an atomic overwrite (write to a temp key, then rename), so re-running a task converges rather than accumulates. The full treatment of atomic geospatial writes and skip-if-exists patterns is in writing idempotent Airflow DAGs for geospatial refresh.
• Schema drift. An upstream provider silently adds, renames, or retypes a column, and a join that used to match on geoid starts producing all-null attributes. The validation gate should assert an expected schema and route mismatches to the dead-letter store rather than letting a structurally-valid-but-semantically-wrong frame flow downstream. Catching this class of regression across refreshes is the subject of monitoring spatial data drift in CI pipelines.
• Retrying the non-retryable. A transient 503 from the routing engine deserves a backoff retry; an “origin not snappable to the graph” error will fail identically forever and should raise AirflowFailException (which bypasses retries) and land in the dead-letter queue. Retrying a permanent error just burns the schedule window.
• CRS drift between tasks. A GeoParquet round-trip that loses CRS metadata makes the spatial join produce planar nonsense. Every stage re-asserts crs on read; the assertion fails loudly rather than computing a wrong-but-plausible area.

Performance and scaling #

The binding constraints are the routing engine’s request capacity and the memory footprint of the geometry, not scheduler overhead. Three levers keep a national run inside its window.

1. Bound concurrency with a pool, not with parallelism. The routing_pool caps total in-flight isochrone tasks across every active run to the engine’s sustainable throughput. Widening max_active_tis_per_dag beyond the pool size just queues tasks; sizing the pool to the engine is what matters.
2. Cache aggressively at the routing layer. Overlapping trade areas across weekly refreshes generate enormous redundancy. Keying catchments by (lon, lat, minutes, profile, graph_version) and checking the cache first eliminates most solves; the caching strategies for repeated network queries patterns apply directly, and cache invalidation on network changes is covered in invalidating isochrone caches on traffic updates.
3. Stream to GeoParquet, do not accumulate. Holding every region’s attributed catchments in one process exhausts RAM. Each mapped task writes its partition and passes a URI; only the final score stage concatenates, and it reads column-pruned partitions rather than whole files.

For the heaviest stages, the DAG should run tasks on a KubernetesExecutor so each mapped instance gets its own right-sized pod — a memory-hungry join and a CPU-light export do not need identical resources.

Validation and QA gates #

A gate is a task that refuses to let the batch proceed unless its checks pass. The refresh runs three, each an explicit stage rather than advisory logging.

1. Input gate (post-ingest). CRS present and correct, coordinates within WGS84 bounds, geometry valid, expected schema present. Failures route to the dead-letter store; the batch continues on the clean remainder.
2. Join-integrity gate (post-join). Assert that the attributed row count equals the catchment row count (a left join must not drop rows) and that the null-attribute rate sits below a threshold — a spike signals the schema drift described above. The accuracy of the join itself is validated against known figures per validating spatial join accuracy with ground truth.
3. Drift gate (pre-export). Compare this run’s key distributions — row counts, bounding box, score histogram — against the previous successful run and fail the export if any metric breaches its tolerance. This is the last line of defense against publishing a subtly-broken refresh, and its mechanics are the whole of monitoring spatial data drift in CI pipelines.

@task
def drift_gate(current_uri: str, key: str) -> str:
    cur = gpd.read_parquet(current_uri)
    prev = load_previous_successful_run()          # None on first run
    if prev is not None:
        cur_km2 = cur.to_crs(EQUAL_AREA_CRS).geometry.area.sum() / 1e6
        prev_km2 = prev.to_crs(EQUAL_AREA_CRS).geometry.area.sum() / 1e6
        delta = abs(cur_km2 - prev_km2) / prev_km2
        if delta > 0.15:                            # >15% catchment-area shift
            raise AirflowFailException(f"catchment area drift {delta:.1%}")
    return current_uri

Integration notes #

Airflow owns the heavy, stateful weekly refresh, but not every job needs that heavy machinery. A lightweight daily freshness check — re-pulling changed ACS rows, revalidating coordinates, publishing a small artifact — runs perfectly well on a scheduled GitHub Actions workflow, which also doubles as the CI that runs the drift checks on every code change before the DAG ever deploys. The division of labor is deliberate: Actions for cron-triggered validation and CI, Airflow for the orchestrated production graph.

Downstream, the scored catchments this DAG exports are the raw material for suitability scoring and site ranking models — the ranked shortlist a real-estate committee argues over. Because every run is content-addressed and every gate is enforced, a number in that shortlist can be traced back through the export, the score, the join, and the isochrone to the exact coordinate snapshot and graph version that produced it. That traceability is the point: orchestration is not about running jobs on a schedule, it is about making the output defensible.

Frequently Asked Questions #

Related #

• Writing Idempotent Airflow DAGs for Geospatial Refresh — deterministic keys and atomic writes that make retries safe.
• GitHub Actions for Scheduled Spatial Data Refresh — cron-triggered lightweight refreshes and CI validation.
• Monitoring Spatial Data Drift in CI Pipelines — the drift gate that guards every export.
• Setting Up PostGIS for Retail Analytics — the transactional store the export stage upserts into.
• Suitability Scoring & Site Ranking Models — the downstream models this refresh feeds.

← Back to Location Intelligence Architecture & Data Foundations