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.

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 POP_2025 to pop_2026 and a column quietly fills with nulls.
Wrong units — a file arrives in degrees instead of metres and every area is nonsense by a factor of 10¹⁰.
Silent row loss — 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.

The goal: every stage states what it requires and what it guarantees, checks run automatically on every run, and a violation stops the pipeline with a message naming exactly which expectation 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.

An input gate before the pipeline and an output gate after it, with assertions between stages. — Two gates and a few assertions catch nearly every silent failure.

import geopandas as gpd

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

gdf = assert_valid(gpd.read_file(cfg["source"]),
                   name="input", required_columns=["parcel_id", "use_class"],
                   crs="EPSG:27700", geom_types=["Polygon", "MultiPolygon"])

Step-by-step solution

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"])
    return assert_valid(
        gdf,
        name=f"input:{cfg['source'].name}",
        required_columns=cfg["expected_columns"],
        crs=cfg["source_crs"],
        min_rows=1,
        geom_types=["Polygon", "MultiPolygon"],
    )

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

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):
    before = len(gdf)
    joined = gpd.sjoin(gdf, cfg["wards"].to_crs(gdf.crs),
                       how="left", predicate="within")

    if len(joined) > before:
        dupes = len(joined) - before
        raise ValidationError(
            f"spatial join produced {dupes} extra rows — "
            "features matched more than one ward (overlapping polygons?)")

    unmatched = 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")

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.

Four silent failures — empty output, wrong CRS, renamed column, duplicated rows — each with the check that catches it. — Each silent failure has one cheap check that turns it into a loud one.

def load(gdf, cfg, features_in):
    problems = check(gdf, name="output", crs=cfg["target_crs"], min_rows=1,
                     required_columns=["parcel_id", "area_m2"])

    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_suffix(".gpkg.tmp")
    gdf.to_file(tmp, driver="GPKG")
    tmp.replace(cfg["out"])            # only a validated result becomes the output
    return cfg["out"]

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. Three invalid geometries out of 40,000 may be worth repairing and 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")

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 understands GeoPandas and validates dtypes, ranges, nullability, and uniqueness in one place.

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

class ParcelSchema(pa.DataFrameModel):
    parcel_id: Series[str] = pa.Field(unique=True, nullable=False)
    use_class: Series[str] = pa.Field(isin=["residential", "commercial", "industrial"])
    area_m2: Series[float] = pa.Field(gt=0, le=1e8)
    year_built: Series[int] = pa.Field(ge=1800, le=2100, nullable=True)

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

validated = ParcelSchema.validate(gdf, lazy=True)   # lazy → report every failure

lazy=True is the same principle as collecting problems in a list: one run tells you everything that 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["validation"] = {
    "input_rows": features_in,
    "output_rows": len(result),
    "retention": round(len(result) / features_in, 4),
    "checks_run": ["crs", "columns", "geometry", "retention"],
    "warnings": [f.message for f in findings if f.severity == "warning"],
}

Code examples

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, field

@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=("parcel_id", "use_class"),
    geom_types=("Polygon", "MultiPolygon"), min_rows=100)

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 "
                                  f"(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

The 10 % rule catches the classic wrong-CRS clip: it does not usually return some features, it returns almost none.

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

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

import rasterio

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]), 6), round(abs(src.res[1]), 6)) != res:
            problems.append(f"{path.name}: resolution {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

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.

import json
from pathlib import Path

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

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.

import pandas as pd

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

validation_report([
    ("input",  "crs",        "pass", "EPSG:27700"),
    ("input",  "columns",    "pass", "parcel_id, use_class"),
    ("clean",  "geometry",   "warn", "12 invalid geometries repaired"),
    ("join",   "row_count",  "pass", "18,204 → 18,204"),
    ("output", "retention",  "pass", "97%"),
], Path("logs/validation.csv"))

Explanation

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.

A checklist of pipeline validations: CRS, required columns, null geometry, geometry types, row counts, retention. — Six checks cover the overwhelming majority of real pipeline failures.

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 eleven rows with null geometry is often correct — but if next month there are eleven thousand, 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 six problems 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 0.3 % of features. A single line asserting that at least half the input survives would have caught it in the first run, before the number reached a report.

Edge cases or notes

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

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.

How to Validate Pipeline Inputs and Outputs Automatically in Python #

Problem statement #

Quick answer #

Step-by-step solution #

Validate the input before doing any work #

State each stage's contract #

Check row counts across every join #

Refuse to write an empty or implausible output #

Distinguish errors from warnings #

Reach for a schema library when the rules multiply #

Record what passed #

Code examples #

Example 1: A reusable expectations object #

Example 2: Validating between every stage automatically #

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

Example 4: Comparing a run against the last one #

Example 5: A validation report humans will read #

Explanation #

Edge cases or notes #

Legitimately empty results #

CRS comparison is subtler than string equality #

Validation cost on large data #

Over-strict schemas break on harmless changes #

Warnings nobody reads #

Validation is not a substitute for tests #

Internal links #

FAQ #

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

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

Should validation raise an exception or log a warning? #

What is a retention check and why does it matter? #

Should I use pandera or write checks by hand? #

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

Does validation slow the pipeline down? #