Location Intelligence Retail Site Selection Automation

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.

Metric Definition Default threshold Catches
Row-count delta abs(n_cur - n_prev) / n_prev > 0.10 Dropped or duplicated records
Bounding-box shift max corner displacement (m, projected) > 5000 Coordinate batch shift, wrong region
CRS change crs_cur != crs_prev any change Dropped or reprojected geometry
Geometry-validity rate share failing is_valid > 0.01 Topology regression from upstream
Demographic PSI population stability index of a joined variable > 0.25 Distribution shift in the join
Null-attribute rate delta change in share of null joined attrs > 0.05 Schema drift, broken join key
Drift has four shapes and only one of them is obvious Volume drift changes the row count, distribution drift moves an attribute's shape without changing the count, spatial drift moves geometry while attributes hold, and schema drift changes the columns. A check that watches only row counts sees the first and misses the other three. Row counts catch one of these four volume rows: 4,610 → 3,902 a truncated delivery, a filter left in place caught by a count check distribution rows unchanged median income +18% a new survey vintage needs a distance metric spatial attributes unchanged centroids shift 400 m a re-projected source needs a geometry check schema a column renamed, a type widened, a code list extended needs a contract test The three on the right are the dangerous ones, because the pipeline keeps running and the numbers keep looking reasonable — the trade areas simply describe a slightly different world than last month. Each needs its own statistic, and each needs a threshold agreed in advance rather than argued about on the morning the alert fires.

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 BB buckets, with aia_i the fraction of the actual (current) snapshot in bucket ii and eie_i the fraction of the expected (baseline) snapshot, PSI is:

PSI=∑i=1B(ai−ei) ln⁡ ⁣(aiei)\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.10.1 as no material shift, 0.10.1–0.250.25 as moderate drift worth a look, and above 0.250.25 as a significant shift that should fail the gate. For a scalar summary of demographic change, a population-weighted percent change across regions rr complements PSI:

Δpop=∑rPr ∣vrcur−vrprev∣∑rPr vrprev\Delta_{pop} = \frac{\sum_{r} P_r \,\lvert v_r^{cur} - v_r^{prev}\rvert}{\sum_{r} P_r \, v_r^{prev}}

where PrP_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; Δpop\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.

python
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.

Twelve refreshes, one real event The drift statistic sits between 0.01 and 0.04 for ten months, spikes to 0.23 in month eleven when a new survey vintage lands, and settles at 0.05 afterwards. The warn line at 0.10 and the fail line at 0.20 separate ordinary month-to-month noise from a change worth stopping the pipeline for. A threshold is only meaningful against a baseline of ordinary months 0.25 0.15 0.05 warn 0.10 fail 0.20 0.23 month 1 month 11 The spike is a new survey vintage — a real, explainable change that should stop the refresh until someone confirms it, not a bug. The following month returning to 0.05 is the evidence that it was absorbed cleanly.

Failure modes and debugging

Symptom Cause Fix
PSI is inf or nan Empty bin gives log(0) or divide-by-zero Clip bin fractions to a small epsilon before the log.
Bounding-box shift always huge Measured in degrees, or CRS mismatch Reproject both snapshots to an equal-area CRS first.
Gate never fires on real drift Baseline is itself a bad run Pin the baseline to a run that passed all gates, not just the last run.
Gate fires on every deliberate update Threshold too tight for a real refresh cadence Set tolerances with data owners; exempt intended graph/vintage bumps.
PSI unstable run to run Too many bins for the sample size Use 10 quantile bins; reduce for small samples.
CRS check passes but areas differ Same EPSG code, different axis/datum assumption Compare via pyproj.CRS equality, not string equality.

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.

Three bands, three behaviours Below the warn threshold the refresh publishes silently. Between warn and fail it publishes and posts a note naming the drifted columns. Above the fail threshold it writes the candidate output to a quarantine location, leaves the previous version serving, and asks for a human decision. The gate publishes, annotates, or quarantines — never guesses below 0.10 · publish the new version becomes current, no message sent the statistic is still recorded so the baseline keeps growing 0.10 to 0.20 · annotate publishes, and posts which columns moved and by how much the note lands beside the dataset version, not only in chat above 0.20 · quarantine candidate written aside, previous version keeps serving approving it is one command and it is recorded as a decision Quarantine rather than failure matters: a failed job leaves nothing to inspect and tempts whoever is on call to rerun it until it passes. A quarantined artifact is the evidence needed to make the decision properly. Thresholds are per dataset; a competitor feed and a census extract do not deserve the same sensitivity.

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.
python
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.

Frequently Asked Questions

What baseline should drift be measured against?

The previous accepted version, plus a longer-run reference such as the same month last year where seasonality matters. Comparing only against the immediately previous run makes a slow shift invisible — each step is small, and the model wanders a long way over a year without ever tripping a threshold. Holding both comparisons costs one extra statistic and catches the class of change that a single-step check is structurally unable to see.

Does drift monitoring belong in CI or in the pipeline?

In the pipeline, with the results surfaced in CI. The check needs the data, which CI often does not have in full, and it needs to run on every scheduled refresh rather than on every commit. What belongs in CI is the test that the drift check itself still works — a fixture with known drift that must be detected, and one without that must not be. That fixture is what stops the check silently degrading into an assertion that never fires.

How should thresholds be chosen?

From the history you already have, not from a rule of thumb. Compute the statistic across the last year of refreshes, look at where ordinary months sit, and set the warn line above that band and the fail line where a change would genuinely alter a decision. A threshold copied from another dataset will either alert every month or never, and both outcomes train the team to ignore it.

← Back to Orchestrating Spatial Pipelines with Airflow