Setting Up Spatial Pipeline Health Checks in Apache Airflow

Spatial ETL pipelines fail differently from tabular ones. A self-intersecting polygon, a silently re-projected feed, or a bloated GiST index does not raise a row-count error — it slips through ingestion and only surfaces hours later when a tile service returns null geometries or misaligned features. This page is the focused implementation guide for one narrow but high-value scenario: wiring deterministic spatial health checks into an Apache Airflow DAG so structural faults fail fast at the ingestion boundary and never reach downstream transforms. It sits under the Geospatial Metric Taxonomy for ETL reference, which defines the gis.spatial.* and gis.etl.* metric names used throughout, and within the broader Geospatial Observability Architecture & Fundamentals domain. The audience is the SREs, GIS platform administrators, and data engineers who own pipeline mean time to resolution (MTTR).

Problem Framing

The scenario this guide targets is the silent structural fault — a geometry-level defect that passes every generic data-quality check because the rows arrive, the columns are populated, and the job exits zero. The defect only becomes visible after it has propagated into a published layer, by which point remediation means reprocessing every downstream artifact rather than skipping a single batch.

Three signal classes indicate the problem is present and where it lives in the pipeline:

• Geometry validity faults (gis.spatial.geometry_invalid_ratio) — self-intersections, ring-ordering errors, and empty geometries originate at the ingest stage, usually from a malformed source extract or a GDAL conversion that lost precision.
• Coordinate reference drift (gis.spatial.crs_mismatch_count) — a feed re-projected or tagged with the wrong SRID enters at ingest but only manifests as misalignment after the transform stage joins it against authoritative layers.
• Index and extent anomalies (gis.spatial.index_utilization_ratio, gis.spatial.extent_drift_meters) — bloated spatial indexes and bounding-box shifts surface at the index and publish stages, degrading query latency long before they error.

Because these faults are stage-specific, the health checks must run as a dedicated gate immediately after raw ingestion and before any transform task. Placing them anywhere later means the corruption has already crossed a spatial data trust boundary and every guarantee downstream tasks rely on is void. Isolate the checks into their own task group with strict concurrency limits so a heavy index rebuild cannot contend for locks with a live validation query.

Implementation

The DAG below establishes the execution contract: bounded retries with exponential backoff, an SLA, single-run concurrency to avoid lock contention, and a failure callback that emits a critical log line for the alert router. Airflow 3.x uses schedule rather than the removed schedule_interval argument.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
from airflow.exceptions import AirflowSkipException
from airflow.providers.postgres.hooks.postgres import PostgresHook
import logging

default_args = {
    "owner": "sre-gis-ops",
    "retries": 2,
    "retry_delay": timedelta(seconds=30),
    "retry_exponential_backoff": True,
    "sla": timedelta(hours=2),
    "on_failure_callback": lambda context: logging.critical(
        f"Spatial health check failed: {context['task_instance']}"
    )
}

dag = DAG(
    dag_id="spatial_ingestion_health_checks",
    default_args=default_args,
    schedule="@daily",  # `schedule` supersedes the removed `schedule_interval` arg in Airflow 3.x
    start_date=datetime(2024, 1, 1),
    tags=["spatial-observability", "mttr-reduction", "postgis"],
    catchup=False,
    max_active_runs=1,   # serialize runs so index rebuilds never overlap a validation query
    max_active_tasks=1,
)

Validity and CRS gate

The first task batches a single deterministic query against the freshly ingested batch, computing the invalid-geometry ratio and the count of distinct SRIDs. The validity ratio is the proportion of features failing ST_IsValid or returning ST_IsEmpty:

$$r_{\text{invalid}} = \frac{|\{g : \lnot\,\text{ST\_IsValid}(g) \lor \text{ST\_IsEmpty}(g)\}|}{N_{\text{features}}}$$

A breach raises AirflowSkipException so downstream consumers are skipped rather than marked failed — the batch is quarantined, not the pipeline. A multi-SRID result instead raises a hard error, because a mixed coordinate reference system cannot be auto-skipped and demands projection realignment.

def validate_spatial_integrity(**kwargs):
    hook = PostgresHook(postgres_conn_id="gis_warehouse")
    conn = hook.get_conn()
    cursor = conn.cursor()

    cursor.execute("""
        SELECT
            COUNT(*) AS total_features,
            SUM(CASE WHEN NOT ST_IsValid(geom) OR ST_IsEmpty(geom) THEN 1 ELSE 0 END) AS invalid_features,
            COUNT(DISTINCT ST_SRID(geom)) AS unique_srids,
            MIN(ST_SRID(geom)) AS expected_srid
        FROM raw_ingestion_layer
        WHERE ingestion_batch_id = %s;
    """, (kwargs['dag_run'].run_id,))

    total, invalid, unique_srids, expected = cursor.fetchone()
    invalid_ratio = invalid / total if total > 0 else 0.0  # gis.spatial.geometry_invalid_ratio

    if invalid_ratio > 0.005:
        raise AirflowSkipException(
            f"Geometry validity threshold breached: {invalid_ratio:.4%} invalid. "
            f"Skipping downstream consumers to prevent topology corruption."
        )

    if unique_srids > 1:  # gis.spatial.crs_mismatch_count > 0
        raise ValueError(
            f"CRS mismatch detected: {unique_srids} distinct SRIDs found. "
            f"Expected uniform {expected}. Halt pipeline for projection realignment."
        )

    logging.info(f"Spatial trust boundaries validated: {total} features, {invalid_ratio:.4%} invalid ratio.")
    cursor.close()
    conn.close()

validate_task = PythonOperator(
    task_id="validate_spatial_integrity",
    python_callable=validate_spatial_integrity,
    dag=dag,
)

Threshold reference and alert routing

Each metric maps to a precise severity tier and routing target so spatial faults bypass the generic SRE queue and reach the team that can act. Use the canonical names from the metric taxonomy so dashboards, alert rules, and DAG logs all refer to the same signal.

The index-utilization check reads PostgreSQL system catalogs directly. A low ratio means sequential scans dominate a table large enough to warrant an index — the signature of a bloated or invalidated GiST index that needs REINDEX CONCURRENTLY.

def check_index_fragmentation(**kwargs):
    hook = PostgresHook(postgres_conn_id="gis_warehouse")
    conn = hook.get_conn()
    cursor = conn.cursor()
    cursor.execute("""
        SELECT
            relname,
            idx_scan,
            seq_scan,
            CASE WHEN idx_scan + seq_scan = 0 THEN 0
                 ELSE idx_scan::float / (idx_scan + seq_scan)
            END AS index_utilization
        FROM pg_stat_user_tables
        WHERE idx_scan + seq_scan > 1000
          AND CASE WHEN idx_scan + seq_scan = 0 THEN 0
                   ELSE idx_scan::float / (idx_scan + seq_scan)
              END < 0.65;
    """)

    degraded_tables = cursor.fetchall()
    if degraded_tables:
        table_names = [row[0] for row in degraded_tables]
        raise RuntimeError(
            f"Spatial index degradation detected: {', '.join(table_names)}. "
            "Initiate REINDEX CONCURRENTLY."
        )
    cursor.close()
    conn.close()

Extent drift detection

Compare the ingested batch’s bounding box against a known-good historical extent. A centroid shift over 5.0 meters or an area ratio outside ±5% almost always indicates a CRS misassignment or a projection-transform error rather than legitimate data growth. Tune these tolerances per feature class — cadastral and parcel layers are far tighter than administrative boundaries, a distinction covered in the observability scoping rules for vector data.

def detect_extent_drift(**kwargs):
    hook = PostgresHook(postgres_conn_id="gis_warehouse")
    conn = hook.get_conn()
    cursor = conn.cursor()

    cursor.execute("""
        WITH current_extent AS (
            SELECT ST_Extent(geom) AS bbox FROM raw_ingestion_layer
            WHERE ingestion_batch_id = %s
        ),
        historical_extent AS (
            SELECT ST_Extent(geom) AS bbox FROM validated_master_layer
            WHERE feature_type = 'cadastral'
        )
        SELECT
            ST_Distance(
                ST_Transform(ST_Centroid(c.bbox::geometry), 3857),
                ST_Transform(ST_Centroid(h.bbox::geometry), 3857)
            ) AS centroid_drift_meters,
            ST_Area(c.bbox::geometry) / NULLIF(ST_Area(h.bbox::geometry), 0) AS area_ratio
        FROM current_extent c, historical_extent h;
    """, (kwargs['dag_run'].run_id,))

    drift_m, area_ratio = cursor.fetchone()  # gis.spatial.extent_drift_meters

    if drift_m > 5.0 or (area_ratio is not None and abs(area_ratio - 1.0) > 0.05):
        raise AirflowSkipException(
            f"Spatial extent drift detected: {drift_m:.2f}m centroid shift, "
            f"{area_ratio:.2f} area ratio. Halt for CRS verification."
        )
    cursor.close()
    conn.close()

Tracing the checks

Wrap each validation in an OpenTelemetry span so check duration and feature counts correlate with downstream GIS API latency. Full collector configuration lives in the OpenTelemetry integration for GIS pipelines guide; the task-level instrumentation is minimal:

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.trace import Status, StatusCode

trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer("spatial-etl-health")

def otel_instrumented_validation(**kwargs):
    with tracer.start_as_current_span("spatial_integrity_check") as span:
        span.set_attribute("spatial.feature_type", "cadastral")
        span.set_attribute("spatial.validation_engine", "postgis")
        try:
            validate_spatial_integrity(**kwargs)
            span.set_status(Status(StatusCode.OK))
        except Exception as e:
            span.set_status(Status(StatusCode.ERROR, str(e)))
            span.record_exception(e)
            raise

Verification & Testing

Confirm the gate works before trusting it in production by proving it both passes clean data and quarantines bad data:

1. Synthetic invalid-geometry injection. Insert a known self-intersecting polygon into a staging copy of raw_ingestion_layer and run the task. The bowtie below pushes gis.spatial.geometry_invalid_ratio above 0.005 and must raise AirflowSkipException, not pass:

   INSERT INTO raw_ingestion_layer (geom, ingestion_batch_id)
   VALUES (ST_GeomFromText('POLYGON((0 0, 1 1, 1 0, 0 1, 0 0))', 4326), 'test-batch-001');

2. CRS mismatch assertion. Add a second row tagged with a different SRID and confirm the task raises a ValueError for projection realignment rather than skipping.

3. Skip-vs-fail semantics. In the Airflow UI, verify the downstream transform tasks show state skipped (yellow) after a validity breach and upstream_failed only after a hard CRS error — they must never run on a quarantined batch.

4. Query-plan check. Run EXPLAIN (ANALYZE, BUFFERS) on the validation query against a production-sized table to confirm it uses the spatial index and stays inside the SLA window. A sequential scan here means the check itself will time out under load.

Gotchas & Failure Modes

SRID mismatch silently passing validity checks. ST_IsValid evaluates a geometry in its own declared coordinate space, so a feature tagged with the wrong SRID can be perfectly valid yet land in the wrong hemisphere. Validity and CRS are independent signals — the COUNT(DISTINCT ST_SRID(geom)) check is what catches this, which is why it must run in the same gate and not be dropped as redundant.

AirflowSkipException versus raise. Using a hard raise for a recoverable validity breach marks the task failed and triggers retries that re-run the same doomed batch, burning the retry budget. Reserve raise for faults that genuinely require human intervention (mixed SRIDs); use AirflowSkipException for quarantine-and-continue so the next scheduled batch proceeds cleanly.

Extent drift false positives on legitimate growth. A genuinely expanding dataset — a new annexation in a cadastral layer — trips the 5% area threshold and blocks a valid batch. Scope the historical comparison to a stable reference subset, or widen the tolerance per feature class, so real growth is not mistaken for projection error.

REINDEX CONCURRENTLY lock contention. Firing the index remediation from inside a concurrent DAG run competes for the same table locks the validation queries hold. The max_active_runs=1 and max_active_tasks=1 settings serialize this; monitor pg_stat_progress_create_index during the rebuild. When primary spatial APIs cannot serve during a rebuild, route reads to cached snapshots via fallback chains for spatial API failures to keep MTTR bounded.

FAQ

Where in the DAG should spatial health checks run?

Immediately after raw ingestion and before any transform, tile, or publish task. Running them later means corruption has already crossed the trust boundary and contaminated downstream artifacts, turning a one-batch skip into a full reprocess.

Should a validity breach fail the task or skip it?

Skip it with AirflowSkipException. That quarantines the bad batch and lets the next scheduled run proceed, instead of marking the task failed, exhausting retries on the same data, and paging on-call. Reserve a hard raise for unrecoverable faults like a mixed-SRID batch.

How do I tune the 0.005 validity threshold?

Start from the observed baseline invalid ratio of clean batches and set the alert at a few standard deviations above it. Source feeds with chronic minor self-intersections may justify a higher gate paired with an automatic ST_MakeValid repair step; authoritative cadastral feeds should sit near zero.

Why check both ST_IsValid and ST_SRID separately?

They detect orthogonal faults. ST_IsValid catches structural defects within a coordinate space; the distinct-SRID count catches a geometry that is structurally perfect but projected wrong. A single check cannot cover both, and the CRS fault is the one that silently misaligns published layers.

What signals confirm a spatial index, not the query, is the bottleneck?

A low gis.spatial.index_utilization_ratio (sequential scans dominating a large table) plus an EXPLAIN (ANALYZE, BUFFERS) plan showing a sequential scan where a GiST index scan is expected. Confirm with pg_stat_user_tables before triggering REINDEX CONCURRENTLY.

Related

• Geospatial Metric Taxonomy for ETL — the parent reference defining the gis.spatial.* and gis.etl.* metric names these checks emit.
• Defining Spatial Data Trust Boundaries — where in the pipeline a geometry becomes trusted, and why the gate belongs at ingestion.
• OpenTelemetry Integration for GIS Pipelines — collector configuration for shipping the spans these tasks open.
• Fallback Chains for Spatial API Failures — keeping reads available while a flagged batch or index rebuild is remediated.
• Geospatial Observability Architecture & Fundamentals — the architectural reference for the whole domain.