Monitoring Spatial Data Drift in CI Pipelines

This page solves one exact task: computing drift metrics between two spatial data snapshots in a CI pipeline and failing the build when any metric breaches its threshold, so a subtly-broken refresh never reaches a downstream model.

Drift is change you did not intend. A provider re-tiles their block groups, a geocoder shifts a batch of coordinates, a CRS gets dropped on a round-trip, and the refresh still runs green — the row counts look plausible, the geometries are valid, and the corruption surfaces weeks later as a wrong site score. The defense is an automated comparison of each new snapshot against the last known-good one, run as a gate that fails CI on a breach. This is the drift check the orchestrating spatial pipelines with Airflow refresh runs before every export.

Prerequisites #

Before wiring up drift detection you need:

• Two comparable snapshots — the current refresh output and the previous successful run’s output — both as GeoParquet in the same schema.
• Python packages: geopandas, shapely, pyproj, pandas, and numpy for the distribution math.
• A baseline you trust. The “previous” snapshot must be a run that passed all gates, including the accuracy checks in validating spatial join accuracy with ground truth; comparing against a bad baseline just hides drift.
• Thresholds agreed with the data owners. A 15% catchment-area shift may be alarming for a mature market and normal after a deliberate graph re-extract — the tolerances are a business decision, not a default.

Configuration and execution parameters #

The table lists the drift metrics the gate computes and the tolerance that trips a failure. Each is a distinct failure signature.

The drift statistic #

Row counts and bounding boxes catch structural breaks, but a distribution can shift while the count stays fixed — a demographic join that starts pulling from a re-tiled geography attributes systematically different populations without changing how many rows exist. The Population Stability Index (PSI) quantifies that shift. Binning a variable into $B$ buckets, with $a_i$ the fraction of the actual (current) snapshot in bucket $i$ and $e_i$ the fraction of the expected (baseline) snapshot, PSI is:

$$\mathrm{PSI} = \sum_{i=1}^{B} (a_i - e_i)\,\ln\!\left(\frac{a_i}{e_i}\right)$$

The convention reads a value below $0.1$ as no material shift, $0.1$–$0.25$ as moderate drift worth a look, and above $0.25$ as a significant shift that should fail the gate. For a scalar summary of demographic change, a population-weighted percent change across regions $r$ complements PSI:

$$\Delta_{pop} = \frac{\sum_{r} P_r \,\lvert v_r^{cur} - v_r^{prev}\rvert}{\sum_{r} P_r \, v_r^{prev}}$$

where $P_r$ weights each region by its population so a swing in a dense metro counts more than the same swing in a rural county. PSI flags shape change in a distribution; $\Delta_{pop}$ flags magnitude change weighted by where people actually are.

Annotated implementation #

The function below computes the full drift report between two GeoDataFrame snapshots and raises when any metric breaches its threshold. Every geometric comparison is CRS-aware: the bounding-box shift is measured in a projected equal-area CRS, never in degrees, and a CRS mismatch is itself a drift signal.

import numpy as np
import geopandas as gpd
from pyproj import CRS
from dataclasses import dataclass

STORAGE_CRS = CRS.from_epsg(4326)
EQUAL_AREA_CRS = CRS.from_epsg(5070)   # Albers metres, for displacement + area


@dataclass
class DriftThresholds:
    row_count: float = 0.10
    bbox_m: float = 5000.0
    validity_rate: float = 0.01
    psi: float = 0.25
    null_rate: float = 0.05


def population_stability_index(expected: np.ndarray, actual: np.ndarray,
                               bins: int = 10) -> float:
    """PSI between a baseline and current distribution of one variable."""
    edges = np.quantile(expected, np.linspace(0, 1, bins + 1))
    edges[0], edges[-1] = -np.inf, np.inf          # open outer bins
    e = np.histogram(expected, edges)[0] / len(expected)
    a = np.histogram(actual, edges)[0] / len(actual)
    eps = 1e-6                                       # avoid log(0) / divide-by-0
    e, a = np.clip(e, eps, None), np.clip(a, eps, None)
    return float(np.sum((a - e) * np.log(a / e)))


def detect_drift(prev: gpd.GeoDataFrame, cur: gpd.GeoDataFrame,
                 variable: str, th: DriftThresholds) -> dict:
    report, breaches = {}, []

    # 1. CRS change is itself drift — check before any geometric comparison.
    if prev.crs is None or cur.crs is None:
        raise ValueError("snapshot missing CRS; cannot compare")
    if CRS.from_user_input(prev.crs) != CRS.from_user_input(cur.crs):
        breaches.append(f"CRS changed {prev.crs} -> {cur.crs}")
    prev = prev.to_crs(STORAGE_CRS)
    cur = cur.to_crs(STORAGE_CRS)

    # 2. Row-count delta.
    rc = abs(len(cur) - len(prev)) / max(len(prev), 1)
    report["row_count_delta"] = rc
    if rc > th.row_count:
        breaches.append(f"row-count delta {rc:.1%}")

    # 3. Bounding-box shift, measured in projected metres.
    pb = prev.to_crs(EQUAL_AREA_CRS).total_bounds
    cb = cur.to_crs(EQUAL_AREA_CRS).total_bounds
    shift = float(np.abs(cb - pb).max())
    report["bbox_shift_m"] = shift
    if shift > th.bbox_m:
        breaches.append(f"bbox shift {shift:.0f} m")

    # 4. Geometry-validity regression.
    vr = float((~cur.geometry.is_valid).mean())
    report["invalid_rate"] = vr
    if vr > th.validity_rate:
        breaches.append(f"invalid geometry rate {vr:.1%}")

    # 5. Null-attribute rate delta on the joined variable.
    nr = float(cur[variable].isna().mean() - prev[variable].isna().mean())
    report["null_rate_delta"] = nr
    if nr > th.null_rate:
        breaches.append(f"null-attribute rate rose {nr:.1%}")

    # 6. Distribution shift via PSI on the joined variable.
    psi = population_stability_index(
        prev[variable].dropna().to_numpy(),
        cur[variable].dropna().to_numpy(),
    )
    report["psi"] = psi
    if psi > th.psi:
        breaches.append(f"PSI {psi:.3f} exceeds {th.psi}")

    report["breaches"] = breaches
    if breaches:
        raise DriftError("; ".join(breaches))       # fail CI
    return report


class DriftError(Exception):
    """Raised to fail the CI pipeline on a threshold breach."""

Wired into CI, detect_drift runs as the last gate before publish: a raised DriftError returns a non-zero exit code, the workflow step fails, and the export never happens. The coordinate-level validity checks it leans on are the same ones defined in data validation rules for store coordinates — drift monitoring extends single-snapshot validation into a between-snapshot comparison.

Failure modes and debugging #

The trap is an unstable baseline. Drift is defined relative to a reference, so if the reference is a run that itself drifted, the gate silently normalizes the corruption and stops catching it. Anchor the baseline to a run that cleared every gate — including the ground-truth accuracy check — and only advance it deliberately.

Verification #

Prove the gate works before trusting it to guard production.

1. Alert fires on synthetic drift. Perturb a copy of the current snapshot — shift 20% of a variable’s values, or nudge a batch of coordinates 10 km — and confirm detect_drift raises. A gate that never fails is untested, not healthy.
2. No false alarm on an identical snapshot. Run the gate with prev == cur and confirm PSI is near zero and no breach is raised.
3. CRS regression is caught. Feed a snapshot reprojected to a different CRS and confirm the CRS-change breach triggers.
4. Thresholds are exercised. Push a metric just over and just under its tolerance and confirm the gate flips exactly at the boundary.

import numpy as np

# synthetic drift: shift 20% of median-income values up by 40%
drifted = cur.copy()
idx = drifted.sample(frac=0.2, random_state=0).index
drifted.loc[idx, "median_income"] *= 1.4

try:
    detect_drift(cur, drifted, variable="median_income", th=DriftThresholds())
    raise AssertionError("gate failed to fire on synthetic drift")
except DriftError as e:
    print(f"gate correctly failed: {e}")     # PSI breach expected

A gate that fires on injected drift and stays quiet on an identical snapshot is a gate you can trust to fail the build — and failing the build is the whole point, because a red pipeline is cheap and a silently corrupted site score is not.

Related #

• Orchestrating Spatial Pipelines with Airflow — the refresh whose export this drift gate guards.
• Data Validation Rules for Store Coordinates — the single-snapshot checks drift monitoring extends across runs.
• Validating Spatial Join Accuracy with Ground Truth — how to establish a baseline worth comparing against.

← Back to Orchestrating Spatial Pipelines with Airflow