How to Validate Pipeline Inputs and Outputs Automatically in Python

The worst failure a GIS pipeline can have is not a crash. A crash is honest — it stops, it leaves a traceback, and somebody fixes it. The worst failure is the run that finishes cleanly, writes a file, logs "done", and produces a layer with 0 features because a boundary arrived in the wrong CRS and the clip matched nothing. Nobody notices for three weeks, by which point four reports have quoted the number. Automated validation is how you make that class of failure impossible: explicit checks at the boundaries of every stage, so wrong data stops the pipeline instead of flowing through it.

This page builds those checks around a real pipeline: HM Land Registry's 21,253 INSPIRE parcels for Rutland, measured and joined to the county's 15 ONS wards. The opening story is not hypothetical — the ONS service hands out the same wards in WGS 84 if you ask it to, and a clip against them does this:

with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always")
    clipped = gpd.clip(parcels, wards_wgs84)
print(f"{len(parcels):,} parcels clipped to Rutland's wards: {len(clipped)} rows")
print("warning:", str(caught[0].message).splitlines()[0])

21,253 parcels clipped to Rutland's wards: 0 rows
warning: CRS mismatch between the CRS of left geometries and the CRS of right geometries.

One warning, zero rows, and to_file() will write the empty result without complaint.

Problem statement

Your pipeline runs without errors but you cannot trust the result. The specific fears:

• Empty output, no error — a clip or a join silently matches nothing and writes an empty layer.
• Schema drift — the supplier renames a column and a join key quietly fills with nulls.
• Wrong units — a file arrives in degrees instead of metres and every area is nonsense by a factor of about 10¹⁰.
• Silent row loss or gain — a join duplicates or drops features and the count is never checked.
• Bad geometry passed downstream — invalid polygons survive the cleaning stage and break an overlay two stages later.

Each stage should state what it needs and what it promises, check both on every run, and stop with a message naming the expectation that failed.

Quick answer

Put a validation gate before the first stage and after the last, and assert the critical invariants between stages. Express each check as a small function returning a list of problems, so one run reports every violation instead of only the first.

class ValidationError(Exception):
    pass


def check(gdf, *, name, required_columns=(), crs=None, min_rows=1,
          allow_null_geometry=False, geom_types=None):
    """Return a list of problems — empty means the frame is acceptable."""
    problems = []
    if len(gdf) < min_rows:
        problems.append(f"{name}: expected at least {min_rows} rows, got {len(gdf)}")
    missing = [c for c in required_columns if c not in gdf.columns]
    if missing:
        problems.append(f"{name}: missing columns {missing}")
    if crs is not None:
        if gdf.crs is None:
            problems.append(f"{name}: no CRS set (expected {crs})")
        elif not gdf.crs.equals(crs):
            problems.append(f"{name}: CRS is {gdf.crs.to_string()}, expected {crs}")
    if not allow_null_geometry:
        n_null = int(gdf.geometry.isna().sum() + gdf.geometry.is_empty.sum())
        if n_null:
            problems.append(f"{name}: {n_null} null or empty geometries")
    n_invalid = int((~gdf.geometry.is_valid).sum())
    if n_invalid:
        problems.append(f"{name}: {n_invalid} invalid geometries")
    if geom_types:
        unexpected = set(gdf.geom_type.unique()) - set(geom_types)
        if unexpected:
            problems.append(f"{name}: unexpected geometry types {sorted(unexpected)}")
    return problems


def assert_valid(gdf, **kwargs):
    problems = check(gdf, **kwargs)
    if problems:
        raise ValidationError("Validation failed:\n  - " + "\n  - ".join(problems))
    return gdf

for frame, label in ((wards, "wards.geojson"), (wards_wgs84, "wards_wgs84.geojson"), (clipped, "clip result")):
    try:
        assert_valid(frame, name=label, required_columns=["WD24NM"] if "clip" not in label else ["INSPIREID"],
                     crs="EPSG:27700", geom_types=["Polygon", "MultiPolygon"])
        print(f"{label}: passed")
    except ValidationError as exc:
        print(exc)

wards.geojson: passed
Validation failed:
  - wards_wgs84.geojson: CRS is EPSG:4326, expected EPSG:27700
Validation failed:
  - clip result: expected at least 1 rows, got 0

The gate turns both halves of the opening story into errors: the wards in the wrong CRS are refused before the clip runs, and the empty clip result is refused before anything writes it.

Check inputs, stage contracts and every join

Validate the input before doing any work

The cheapest check is the one that runs first. Reading a file takes seconds; a full pipeline takes minutes. Every second spent validating at the front saves the whole run when the input is wrong.

def extract(cfg):
    gdf = gpd.read_file(cfg["source"], columns=cfg["expected_columns"], DOWNLOAD_SCHEMA="NO")
    return assert_valid(gdf, name="input:parcels", required_columns=cfg["expected_columns"],
                        crs=cfg["source_crs"], min_rows=1, geom_types=["Polygon", "MultiPolygon"])

parcels_in = extract(cfg)
print(f"input passed: {len(parcels_in):,} parcels, {parcels_in.crs.to_string()}")

input passed: 21,253 parcels, EPSG:27700

Note what this catches that a bare read does not: a supplier who dropped a column, a file exported in the wrong projection, an empty extract, and a polygon layer that has quietly acquired point features.

State each stage's contract

A stage has a precondition (what it needs) and a postcondition (what it promises). Writing them down converts assumptions in your head into checks in the code, and the error message becomes the documentation.

def measure_areas(gdf, cfg):
    # precondition: a projected CRS, or "area" is meaningless
    if gdf.crs is None or not gdf.crs.is_projected:
        raise ValidationError(
            f"measure_areas needs a projected CRS, got {gdf.crs}. "
            "Reproject with to_crs('EPSG:27700') before measuring.")
    out = gdf.copy()
    out["area_m2"] = out.geometry.area
    # postcondition: every area is a positive, finite number
    bad = out["area_m2"].isna() | (out["area_m2"] <= 0)
    if bad.any():
        raise ValidationError(f"measure_areas produced {int(bad.sum())} non-positive areas")
    return out

measured = measure_areas(parcels_in, cfg)
print(f"areas: {measured['area_m2'].min():.1f} to {measured['area_m2'].max():,.0f} m²")
try:
    measure_areas(wards_wgs84, cfg)
except ValidationError as exc:
    print(exc)

areas: 0.1 to 7,149,834 m²
measure_areas needs a projected CRS, got EPSG:4326. Reproject with to_crs('EPSG:27700') before measuring.

Without the precondition the wards in degrees would have been "measured" at a few hundredths of a square degree each, and nothing downstream would have known.

Check row counts across every join

Joins are where features vanish and multiply. A spatial join with how="left" should return exactly as many rows as it started with — unless a feature matched twice, which is the single most common surprise in GIS analysis.

def join_wards(gdf, cfg, predicate="within"):
    before = len(gdf)
    joined = gpd.sjoin(gdf, cfg["wards"].to_crs(gdf.crs)[["WD24NM", "geometry"]],
                       how="left", predicate=predicate)
    if len(joined) > before:
        raise ValidationError(
            f"spatial join produced {len(joined) - before} extra rows — "
            "features matched more than one ward")
    unmatched = int(joined["index_right"].isna().sum())
    if unmatched > cfg.get("max_unmatched", 0):
        raise ValidationError(f"{unmatched}/{before} features matched no ward — check CRS and coverage")
    return joined.drop(columns="index_right")

for predicate in ("within", "intersects"):
    try:
        join_wards(measured, cfg, predicate)
    except ValidationError as exc:
        print(f"{predicate:10} {exc}")

within     978/21253 features matched no ward — check CRS and coverage
intersects spatial join produced 597 extra rows — features matched more than one ward

Both versions fail, for opposite reasons, and both failures are real facts about the data. With within, every parcel that straddles a ward boundary matches no ward. With intersects, each of those parcels matches two wards and adds a row. Neither is what the analysis wanted — one ward per parcel — and without the checks the first would have lost rows and the second double-counted area. The fix is to join on a point that lies inside each parcel:

def join_wards_by_point(gdf, cfg):
    points = gdf.set_geometry(gdf.representative_point())
    joined = join_wards(points, {**cfg, "max_unmatched": cfg["max_outside_county"]}, "within")
    joined = joined.set_geometry(gdf.geometry.loc[joined.index]).set_crs(gdf.crs, allow_override=True)
    return joined[joined["WD24NM"].notna()], int(joined["WD24NM"].isna().sum())

with_wards, outside = join_wards_by_point(measured, cfg)
print(f"{len(with_wards):,} parcels, each in exactly one ward; {with_wards['WD24NM'].nunique()} wards used; "
      f"{outside} outside the county")

21,007 parcels, each in exactly one ward; 15 wards used; 246 outside the county

The point join still leaves 246 parcels in no ward, and the check was right to count them. Their points lie outside Rutland — by a median of 46 m — because Land Registry's file for Rutland includes every parcel that touches the county, and most of these lie almost entirely in Leicestershire or Lincolnshire. That is a known property of the source, so the configuration allows up to 300 and the stage drops them and reports how many. When the number jumps, the check fails again.

Refuse to write an empty or implausible output

The output gate is the one that would have caught the story in the opening paragraph. Check that the result exists, is non-empty, is in the CRS you promised downstream, and has not lost an unreasonable share of its features.

def load(gdf, cfg, features_in):
    problems = check(gdf, name="output", crs=cfg["target_crs"], min_rows=1,
                     required_columns=["INSPIREID", "area_m2", "WD24NM"])
    retained = len(gdf) / features_in if features_in else 0
    if retained < cfg.get("min_retention", 0.5):
        problems.append(
            f"output kept only {retained:.0%} of input features "
            f"({len(gdf)}/{features_in}) — below the {cfg['min_retention']:.0%} floor")
    if problems:
        raise ValidationError("Refusing to write:\n  - " + "\n  - ".join(problems))
    tmp = cfg["out"].with_name(cfg["out"].stem + ".partial.gpkg")
    gdf.to_file(tmp, driver="GPKG", layer=cfg["out"].stem)
    tmp.replace(cfg["out"])            # only a validated result becomes the output
    return cfg["out"]

try:
    load(clipped, cfg, len(parcels_in))
except ValidationError as exc:
    print(exc)
print("written:", load(with_wards, cfg, len(parcels_in)))

Refusing to write:
  - output: expected at least 1 rows, got 0
  - output: missing columns ['area_m2', 'WD24NM']
  - output kept only 0% of input features (0/21253) — below the 50% floor
written: data/parcels_by_ward.gpkg

Writing to a temporary file and renaming only after validation passes means a failed run never replaces a good output with a bad one.

Distinguish errors from warnings

Not everything deserves to stop the run. A handful of parcels outside every ward may be worth logging; a wrong CRS is always fatal. Give each check a severity and let the runner decide.

from dataclasses import dataclass


@dataclass
class Finding:
    severity: str      # "error" | "warning"
    message: str


def evaluate(findings, log):
    errors = [f for f in findings if f.severity == "error"]
    for f in findings:
        (log.error if f.severity == "error" else log.warning)(f.message)
    if errors:
        raise ValidationError(f"{len(errors)} blocking problem(s); see the log")


logging.basicConfig(format="%(levelname)s %(message)s", level=logging.INFO, stream=sys.stdout, force=True)
log = logging.getLogger("pipeline")
findings = [Finding("warning", f"{int((with_wards['area_m2'] < 1).sum())} parcels under 1 m²")]
evaluate(findings, log)
try:
    evaluate(findings + [Finding("error", p) for p in check(wards_wgs84, name="wards", crs="EPSG:27700")], log)
except ValidationError as exc:
    print("stopped:", exc)

WARNING 34 parcels under 1 m²
WARNING 34 parcels under 1 m²
ERROR wards: CRS is EPSG:4326, expected EPSG:27700
stopped: 1 blocking problem(s); see the log

Reach for a schema library when the rules multiply

Once you have more than a dozen column-level rules, a declarative schema is easier to read and to maintain than hand-written checks. pandera validates dtypes, ranges, nullability, uniqueness and membership in one place:

import pandera.pandas as pa
from pandera.typing import Series


class ParcelSchema(pa.DataFrameModel):
    INSPIREID: Series[int] = pa.Field(unique=True, nullable=False)
    area_m2: Series[float] = pa.Field(gt=1, le=1e7)
    WD24NM: Series[str] = pa.Field(isin=wards["WD24NM"].tolist())

    class Config:
        strict = False          # allow extra columns
        coerce = True           # cast where it is safe to do so


try:
    ParcelSchema.validate(with_wards, lazy=True)          # lazy → report every failure
except pa.errors.SchemaErrors as exc:
    print(exc.failure_cases[["column", "check", "failure_case"]].groupby(["column", "check"]).size().to_string())

column   check          
area_m2  greater_than(1)    34

lazy=True is the same principle as collecting problems in a list: one run tells you everything that is wrong. Here the only rule that fails is one that was written without looking — "a parcel is bigger than 1 m²" — and the failure cases are the list of real parcels a person should look at before deciding whether the rule or the data is wrong.

Record what passed

A validation that leaves no trace cannot answer "was the data good last Tuesday?". Write the results into the run's manifest alongside the counts.

manifest = {
    "input_rows": len(parcels_in),
    "output_rows": len(with_wards),
    "retention": round(len(with_wards) / len(parcels_in), 4),
    "checks_run": ["crs", "columns", "geometry", "join_count", "retention"],
    "warnings": [f.message for f in findings if f.severity == "warning"],
}
print(json.dumps(manifest))

{"input_rows": 21253, "output_rows": 21007, "retention": 0.9884, "checks_run": ["crs", "columns", "geometry", "join_count", "retention"], "warnings": ["34 parcels under 1 m\u00b2"]}

Expectations between every stage on Rutland's data

The data used on this page

HM Land Registry's INSPIRE Index Polygons for Rutland (© Crown copyright and database rights; OGL); Rutland's wards from the ONS December 2024 boundaries (OGL), requested twice — in British National Grid and in WGS 84; and the Copernicus GLO-30 DEM tile over Rutland (© DLR e.V. 2010–2014 and © Airbus Defence and Space GmbH 2014–2018, provided under COPERNICUS by the European Union and ESA).

import json
import logging
import sys
import warnings
from pathlib import Path

import geopandas as gpd
import pandas as pd
import rasterio
import requests
from rasterio.windows import from_bounds

Path("data").mkdir(exist_ok=True)
Path("logs").mkdir(exist_ok=True)
if not Path("data/Rutland.zip").exists():
    Path("data/Rutland.zip").write_bytes(requests.get(
        "https://use-land-property-data.service.gov.uk/datasets/inspire/download/Rutland.zip", timeout=300).content)
WARDS = ("https://services1.arcgis.com/ESMARspQHYMw9BZ9/arcgis/rest/services/"
         "Wards_December_2024_Boundaries_UK_BFC/FeatureServer/0/query")
for name, srid in (("wards.geojson", 27700), ("wards_wgs84.geojson", 4326)):
    if not Path(f"data/{name}").exists():
        Path(f"data/{name}").write_bytes(requests.get(WARDS, timeout=120, params={
            "where": "LAD24CD='E06000017'", "outFields": "WD24NM", "outSR": srid, "f": "geojson"}).content)
if not Path("data/dem.tif").exists():
    url = ("/vsicurl/https://copernicus-dem-30m.s3.amazonaws.com/Copernicus_DSM_COG_10_N52_00_W001_00_DEM/"
           "Copernicus_DSM_COG_10_N52_00_W001_00_DEM.tif")
    with rasterio.open(url) as src:
        window = from_bounds(-0.83, 52.52, -0.42, 52.76, src.transform)
        profile = src.profile | {"width": int(window.width), "height": int(window.height),
                                 "transform": src.window_transform(window), "driver": "GTiff"}
        with rasterio.open("data/dem.tif", "w", **profile) as dst:
            dst.write(src.read(1, window=window), 1)

parcels = gpd.read_file("/vsizip/data/Rutland.zip/Land_Registry_Cadastral_Parcels.gml",
                        columns=["INSPIREID"], DOWNLOAD_SCHEMA="NO")
wards = gpd.read_file("data/wards.geojson")
wards_wgs84 = gpd.read_file("data/wards_wgs84.geojson")
cfg = {"source": "/vsizip/data/Rutland.zip/Land_Registry_Cadastral_Parcels.gml",
       "expected_columns": ["INSPIREID"], "source_crs": "EPSG:27700", "target_crs": "EPSG:27700",
       "wards": wards, "out": Path("data/parcels_by_ward.gpkg"), "min_retention": 0.5,
       "max_outside_county": 300}
print(f"{len(parcels):,} parcels, {len(wards)} wards ({wards.crs.to_epsg()} and {wards_wgs84.crs.to_epsg()})")

21,253 parcels, 15 wards (27700 and 4326)

Example 1: A reusable expectations object

Describing expectations as data lets you keep them next to the config rather than scattered through the code.

from dataclasses import dataclass


@dataclass
class Expectations:
    name: str
    crs: str | None = None
    required_columns: tuple = ()
    geom_types: tuple = ()
    min_rows: int = 1
    max_null_fraction: float = 0.0

    def check(self, gdf):
        problems = []
        if len(gdf) < self.min_rows:
            problems.append(f"{self.name}: {len(gdf)} rows < {self.min_rows}")
        if self.crs and (gdf.crs is None or not gdf.crs.equals(self.crs)):
            problems.append(f"{self.name}: CRS {gdf.crs} != {self.crs}")
        for col in self.required_columns:
            if col not in gdf.columns:
                problems.append(f"{self.name}: missing column '{col}'")
            elif gdf[col].isna().mean() > self.max_null_fraction:
                problems.append(f"{self.name}: '{col}' is {gdf[col].isna().mean():.1%} null")
        if self.geom_types:
            extra = set(gdf.geom_type.dropna().unique()) - set(self.geom_types)
            if extra:
                problems.append(f"{self.name}: geometry types {sorted(extra)}")
        return problems


PARCELS_IN = Expectations(name="parcels", crs="EPSG:27700", required_columns=("INSPIREID",),
                          geom_types=("Polygon", "MultiPolygon"), min_rows=100)
print(PARCELS_IN.check(parcels) or "parcels: OK")
naive = gpd.sjoin(parcels, wards, how="left", predicate="within")
print(Expectations(name="joined", required_columns=("WD24NM",)).check(naive))

parcels: OK
["joined: 'WD24NM' is 4.6% null"]

max_null_fraction is what catches a join that half-worked: the naive within join has every column, the right CRS and valid geometry, and a ward name missing from 4.6% of its rows.

Example 2: Validating between every stage automatically

Because the runner already wraps each stage, a single change validates the whole pipeline.

def run(gdf, steps, cfg, log):
    data = gdf
    for step in steps:
        before = len(data)
        data = step(data, cfg)
        if data is None or len(data) == 0:
            raise ValidationError(f"step '{step.__name__}' returned no features (input had {before})")
        if before and len(data) / before < 0.1:
            log.warning("step '%s' dropped %.0f%% of features (%d -> %d)",
                        step.__name__, 100 * (1 - len(data) / before), before, len(data))
    return data


def clip_to_county(gdf, cfg):
    return gpd.clip(gdf, cfg["boundary"])


with warnings.catch_warnings():
    warnings.simplefilter("ignore")
    for boundary, label in ((wards, "BNG wards"), (wards_wgs84, "WGS 84 wards")):
        try:
            out = run(parcels, [measure_areas, clip_to_county], {**cfg, "boundary": boundary}, log)
            print(f"{label}: {len(out):,} parcels")
        except ValidationError as exc:
            print(f"{label}: {exc}")

BNG wards: 21,246 parcels
WGS 84 wards: step 'clip_to_county' returned no features (input had 21253)

The wrong-CRS clip does not usually return some features, it returns none, and the runner says which step did it.

Example 3: Checking a raster's grid, not just its pixels

Raster stages have their own invariants — CRS, resolution, nodata, and band count.

def check_raster(path, *, crs, res=None, bands=1, nodata_required=True):
    problems = []
    with rasterio.open(path) as src:
        if src.crs is None or src.crs.to_string() != crs:
            problems.append(f"{path.name}: CRS {src.crs} != {crs}")
        if bands and src.count != bands:
            problems.append(f"{path.name}: {src.count} bands, expected {bands}")
        if res and (round(abs(src.res[0]), 8), round(abs(src.res[1]), 8)) != res:
            problems.append(f"{path.name}: resolution {tuple(round(r, 8) for r in src.res)} != {res}")
        if nodata_required and src.nodata is None:
            problems.append(f"{path.name}: no nodata value set")
        if src.width == 0 or src.height == 0:
            problems.append(f"{path.name}: zero-sized raster")
    return problems


one_arcsec = round(1 / 3600, 8)
print(check_raster(Path("data/dem.tif"), crs="EPSG:4326", res=(one_arcsec, one_arcsec)))

['dem.tif: resolution (0.00041667, 0.00027778) != (0.00027778, 0.00027778)', 'dem.tif: no nodata value set']

Both findings are real properties of the Copernicus DEM rather than mistakes: north of 50° the tiles are 1.5 arc-seconds wide and 1 tall, so a "30 m, 1 arc-second" expectation is wrong in one direction, and the COG tiles carry no nodata value. The check did its job — it made someone write down what the grid actually is.

Example 4: Comparing a run against the last one

Absolute thresholds miss slow drift. Comparing against the previous run catches the day a supplier's feed halves.

def compare_with_previous(manifest, history=Path("logs/history.jsonl"), tolerance=0.2):
    if not history.exists():
        return []
    lines = history.read_text().strip().splitlines()
    if not lines:
        return []
    previous = json.loads(lines[-1])
    problems = []
    prev_rows = previous.get("output_rows", 0)
    if prev_rows:
        change = abs(manifest["output_rows"] - prev_rows) / prev_rows
        if change > tolerance:
            problems.append(f"output row count changed by {change:.0%} "
                            f"({prev_rows} -> {manifest['output_rows']}) — verify the source")
    return problems


history = Path("logs/history.jsonl")
history.write_text(json.dumps(manifest) + "\n")                      # last run
oakham_only = {**manifest, "output_rows": int(with_wards["WD24NM"].str.startswith("Oakham").sum())}
print("same feed:   ", compare_with_previous(manifest, history))
print("feed shrank: ", compare_with_previous(oakham_only, history))

same feed:    []
feed shrank:  ['output row count changed by 76% (21007 -> 4956) — verify the source']

The second run stands in for a supplier whose extract suddenly covered only Oakham: every absolute check would pass it, and the comparison does not.

Example 5: A validation report humans will read

Numbers in a log are easy to skip. A short table at the end of the run is not — built from the checks that actually ran:

def validation_report(rows, out_path):
    report = pd.DataFrame(rows, columns=["stage", "check", "result", "detail"])
    report.to_csv(out_path, index=False)
    print(report.to_string(index=False))
    failed = report[report["result"] != "pass"]
    if not failed.empty:
        print(f"\n{len(failed)} check(s) did not pass")
    return report


validation_report([
    ("input", "crs", "pass", parcels_in.crs.to_string()),
    ("input", "columns", "pass", ", ".join(cfg["expected_columns"])),
    ("join", "row_count", "pass", f"{len(measured):,} -> {len(with_wards):,}"),
    ("join", "within", "warn", f"{int(naive['index_right'].isna().sum()):,} not within one ward; joined by point"),
    ("join", "coverage", "warn", f"{outside} parcels outside the county, dropped"),
    ("output", "retention", "pass", f"{len(with_wards) / len(parcels_in):.0%}"),
], Path("logs/validation.csv"))

 stage     check result                                   detail
 input       crs   pass                               EPSG:27700
 input   columns   pass                                INSPIREID
  join row_count   pass                         21,253 -> 21,007
  join    within   warn 978 not within one ward; joined by point
  join  coverage   warn  246 parcels outside the county, dropped
output retention   pass                                      99%

2 check(s) did not pass

Distrust the world, not your code

Validation is not about distrusting your code; it is about distrusting the world your code runs in. Your transforms are probably correct — what changes without warning is the data arriving at the front door. A column gets renamed, a supplier switches projections, a boundary file gets clipped by someone else, an export runs before an upstream job finished. None of those are bugs in your pipeline, and all of them produce wrong output unless something checks.

The reason to fail loudly rather than repair quietly is that a repair makes a decision on your behalf and hides that it did. Dropping the rows a join failed to match is often correct — but when they are the parcels that straddle a boundary, silently dropping them is a data loss nobody agreed to. Repair the routine cases in a named cleaning stage where the log records exactly how many rows were affected, and let the validation gate fail when the count crosses a threshold you chose deliberately. The data-cleaning checklist covers what belongs in that stage.

Collecting all problems before raising is a small design choice with an outsized effect on how it feels to use. A validator that raises on the first failure produces a fix-run-fail loop: wrong CRS, fix, rerun, missing column, fix, rerun. One that reports every problem at once lets you fix them in a single pass. The same reasoning applies to pandera's lazy=True and to config validation in YAML-driven pipelines.

Retention checks — comparing output count to input count — deserve special mention because they catch the failures that type checks cannot. Every column can have the right dtype, every geometry can be valid, and the answer can still be wrong because a clip in the wrong CRS matched nothing. A single line asserting that at least half the input survives catches it in the first run, before the number reaches a report.

Empty results, CRS equality and over-strict schemas

Legitimately empty results

Sometimes zero features is the correct answer — no incidents this week, no parcels in that category. Make emptiness a configurable expectation (min_rows: 0) rather than removing the check, and log it prominently either way. An empty result that was expected is fine; an empty result nobody looked at is not.

CRS comparison is subtler than string equality

EPSG:4326 and an equivalent WKT definition describe the same system but differ as strings. Compare with gdf.crs.equals(other) from pyproj rather than str(gdf.crs) == "EPSG:4326". Conversely, two CRSs can share a name and differ in datum — see CRS mismatch in GeoPandas.

Validation cost on large data

is_valid runs a full topology check on every geometry and is genuinely slow on millions of features. Validate a sample in-line and run the full check at the gates only, or check validity once after cleaning and record the result rather than re-testing at every stage.

Over-strict schemas break on harmless changes

A schema with strict=True fails when a supplier adds a useful new column. Require the columns you consume, constrain their types and ranges, and allow extras. Strictness belongs on the data you use, not on the shape of the whole file.

Warnings nobody reads

A run that emits forty warnings every time trains everyone to ignore all of them. If a warning has fired on every run for a month, either it should be an error or the condition should be fixed. Keep the warning list short enough that a new entry stands out.

Validation is not a substitute for tests

Checks confirm that this data meets expectations; tests confirm that your code does what you think. You need both — a pipeline can validate perfectly and still compute the wrong thing. Test transform steps on tiny hand-built frames, as described in chaining processing steps.

Internal links

• How to Build a GIS Data Pipeline in Python
• Chain GIS Processing Steps into a Reusable Pipeline
• Drive a GIS Pipeline from a YAML Config File
• Get Alerted When an Automated GIS Job Fails
• Make a GIS Workflow Reproducible
• The Python GIS Data Cleaning Checklist
• How to Validate a GeoDataFrame Against a Schema
• How to Fix CRS Mismatch in GeoPandas
• How to checksum spatial datasets so you can prove they match — content hashes as a pipeline check

FAQ

What should I validate at the start of a GIS pipeline?

The things that make everything downstream wrong: the CRS, the presence and type of the columns you use, that the frame is not empty, that geometries are non-null and valid, and that the geometry type is what your logic assumes. Those five checks take a few milliseconds and catch most bad inputs before a single transform runs.

How do I catch an empty result before it becomes an output file?

Check the row count in the load stage and refuse to write below a threshold, and write to a temporary path that is renamed into place only after the check passes. That way a bad run fails visibly and, crucially, leaves the previous good output intact rather than replacing it with an empty file.

Should validation raise an exception or log a warning?

Both, depending on severity. A wrong CRS, a missing column, or an empty output should raise — continuing produces garbage. A handful of invalid geometries or a few nulls in an optional column should warn, be repaired in a named stage, and be counted. Give every check an explicit severity rather than deciding case by case.

What is a retention check and why does it matter?

It compares the output feature count to the input count and fails if the ratio falls below a floor you set. It catches the failure mode type checks cannot see: a clip or join in the wrong CRS that matches almost nothing. A single assertion that at least half the features survive would prevent most silent GIS pipeline disasters.

Should I use pandera or write checks by hand?

Hand-written checks are perfect for the half-dozen structural rules every pipeline needs — CRS, row count, geometry validity. Reach for pandera when you have many column-level rules (types, ranges, uniqueness, allowed values), because a declarative schema stays readable where a hundred if statements do not. They coexist happily.

How do I validate that a join did not duplicate rows?

Record the row count before the join and compare after. A left join must not increase it; if it does, a feature matched more than one polygon — usually overlapping boundaries. Also count how many rows matched nothing, and fail when that exceeds a threshold. See spatial join returns empty results for the related failure.

Does validation slow the pipeline down?

Structural checks — row counts, column presence, CRS — are effectively free. The one genuinely expensive check is geometry validity on large datasets, because it runs a full topology test per feature. Run that at the gates rather than between every stage, or check a sample in-line and the full set once after cleaning.