What Is a GIS Data Pipeline? The Anatomy Explained

Problem statement

Most GIS automation starts as a script that works and ends as a script nobody dares touch:

# monthly_update.py — 700 lines, no functions
gdf = gpd.read_file("/home/anna/data/parcels_2026_08_FINAL.shp")
gdf = gdf[gdf["STATUS"] != "X"]
# ... 300 lines of transformation, three hard-coded paths, two magic numbers ...
gdf.to_file("/home/anna/outputs/out.shp")
print("done")

It runs on Anna's laptop, on the second Tuesday, if she remembers. It cannot be re-run safely, nobody knows what "STATUS != X" means, and when it produces a wrong number there is no way to find out why.

A pipeline is what that script becomes when you give it structure. The word gets used loosely, so it is worth being precise: a pipeline is a sequence of well-defined steps, driven by configuration, that transforms declared inputs into declared outputs, reproducibly and without a human in the loop.

Quick answer

A GIS pipeline has six parts, and each is a separate concern:

Sources — where data comes from, and how you know it changed
Steps — small, testable transformations with one input and one output
Configuration — paths, parameters and thresholds, outside the code
Orchestration — what runs when, in what order, and what happens on failure
Outputs — deliverables written atomically, with a schema you asserted
Observability — logs, run records, metrics and alerts

from pathlib import Path
import geopandas as gpd

def extract(config) -> gpd.GeoDataFrame:
    return gpd.read_file(config["input"])

def clean(gdf: gpd.GeoDataFrame, config) -> gpd.GeoDataFrame:
    out = gdf[gdf.geometry.notna() & gdf.geometry.is_valid].copy()
    return out.to_crs(config["crs"])

def enrich(gdf: gpd.GeoDataFrame, config) -> gpd.GeoDataFrame:
    out = gdf.copy()
    out["area_m2"] = out.to_crs(out.estimate_utm_crs()).area
    return out

def load(gdf: gpd.GeoDataFrame, config) -> Path:
    dest = Path(config["output"])
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_suffix(dest.suffix + ".part")
    gdf.to_file(tmp, driver="GPKG")
    tmp.replace(dest)                       # atomic: no half-written deliverable
    return dest

def run(config) -> dict:
    gdf = extract(config)
    for step in (clean, enrich):
        before = len(gdf)
        gdf = step(gdf, config)
        print(f"{step.__name__:10} {before} → {len(gdf)} features")
    return {"output": str(load(gdf, config)), "features": len(gdf)}

Four functions and a driver. Each step takes a GeoDataFrame and returns one, which makes every step independently testable and the whole thing readable in a minute.

The anatomy

Layered anatomy of a pipeline: sources, steps, configuration, orchestration, outputs, observability. — Six concerns — a script conflates them, a pipeline separates them.

Step-by-step solution

Vertical steps from a script through extracted functions, config, contracts, orchestration to observability. — The path from working script to dependable pipeline — in the order that pays off soonest.

Steps: one job, one function, no I/O

The single most valuable change is separating transformation from input and output.

import geopandas as gpd

# an impure step: reads, transforms and writes — untestable without files
def bad_clip(input_path, boundary_path, output_path):
    gdf = gpd.read_file(input_path)
    boundary = gpd.read_file(boundary_path)
    gpd.clip(gdf, boundary).to_file(output_path)

# a pure step: frame in, frame out — testable with four synthetic polygons
def clip_to(gdf: gpd.GeoDataFrame, boundary: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
    if gdf.crs != boundary.crs:
        boundary = boundary.to_crs(gdf.crs)
    return gpd.clip(gdf, boundary, keep_geom_type=True)

Pure steps compose, and they can be reasoned about. The impure shell — reading and writing — belongs in one place at the edges.

Contracts: what each step promises

A step that documents and enforces its expectations fails at the right moment, with the right message.

import geopandas as gpd

def requires(gdf: gpd.GeoDataFrame, *, columns=(), crs=None, non_empty=True, label=""):
    missing = [c for c in columns if c not in gdf.columns]
    if missing:
        raise ValueError(f"{label}: missing columns {missing}")
    if crs and gdf.crs != crs:
        raise ValueError(f"{label}: expected {crs}, got {gdf.crs}")
    if non_empty and gdf.empty:
        raise ValueError(f"{label}: input is empty")
    return gdf

def enrich(gdf, config):
    requires(gdf, columns=("parcel_id", "class"), crs=config["crs"], label="enrich")
    out = gdf.copy()
    out["area_m2"] = out.to_crs(out.estimate_utm_crs()).area
    assert (out["area_m2"] > 0).all(), "enrich produced non-positive areas"
    return out

Pre-conditions catch a bad hand-off between steps; post-conditions catch a step that ran but produced nonsense. Both turn a wrong number three stages later into an immediate, located failure.

Configuration: everything that changes without the logic changing

# configs/monthly.yml
input: data/raw/parcels.gpkg
boundary: data/ref/city.gpkg
output: data/out/parcels_clean.gpkg
crs: "EPSG:27700"
min_area_m2: 5
drop_classes: ["exempt", "unknown"]

from pathlib import Path
import yaml

def load_config(path) -> dict:
    path = Path(path).resolve()
    config = yaml.safe_load(path.read_text(encoding="utf-8"))
    for key in ("input", "boundary", "output"):
        config[key] = (path.parent / config[key]).resolve()
    return config

The dividing line is worth stating explicitly: what the pipeline does is code; which data, where, and with what thresholds is configuration. A new month means a new config, never an edited script.

Composition: chaining steps so the pipeline is data

from dataclasses import dataclass
from typing import Callable
import time

@dataclass
class Step:
    name: str
    fn: Callable
    params: dict = None

def run_pipeline(gdf, steps: list[Step], config: dict) -> tuple:
    log = []
    for step in steps:
        started, before = time.perf_counter(), len(gdf)
        gdf = step.fn(gdf, **(step.params or {}))
        log.append({
            "step": step.name,
            "rows_in": before,
            "rows_out": len(gdf),
            "dropped": before - len(gdf),
            "seconds": round(time.perf_counter() - started, 2),
        })
        print(f"{step.name:16} {before:>7,} → {len(gdf):>7,}  ({log[-1]['seconds']}s)")
    return gdf, log

STEPS = [
    Step("drop_invalid", drop_invalid),
    Step("reproject", reproject, {"crs": "EPSG:27700"}),
    Step("clip", clip_to, {"boundary": boundary}),
    Step("enrich", enrich),
]
result, log = run_pipeline(gpd.read_file(config["input"]), STEPS, config)

Once the pipeline is a list, it can be printed, tested, reordered, filtered to a subset for debugging, and — importantly — logged step by step, which is where the row-count ledger comes from.

Idempotency: safe to run twice

from pathlib import Path

def load(gdf, dest: Path) -> Path:
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_suffix(dest.suffix + ".part")
    gdf.to_file(tmp, driver="GPKG")
    tmp.replace(dest)                    # replaces wholesale — no accumulation
    return dest

Re-running with the same inputs must produce the same outputs, not double them. That means overwriting rather than appending, deriving output names deterministically, and never depending on "now" unless the run date is an explicit input.

Orchestration: what runs it, and what happens when it fails

import logging, sys

def main() -> int:
    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(levelname)-7s %(message)s")
    log = logging.getLogger("pipeline")
    try:
        config = load_config(sys.argv[1])
    except Exception as exc:
        log.error("config error: %s", exc)
        return 2                          # refuse to start

    try:
        summary = run(config)
    except Exception:
        log.exception("pipeline failed")
        return 1                          # ran and failed
    log.info("ok: %s", summary)
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

Orchestration is a spectrum: cron running a script, a systemd timer, a GitHub Actions schedule, or Airflow/Prefect/Dagster for a graph of interdependent jobs. The right level is the smallest one that expresses your dependencies — most GIS pipelines never need more than a timer and an exit code.

Observability: the part that makes it maintainable

import json, time
from datetime import datetime, timezone
from pathlib import Path

def write_run_record(summary, step_log, config, dest=Path("logs/runs")):
    dest.mkdir(parents=True, exist_ok=True)
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    record = {
        "run_id": run_id,
        "config": {k: str(v) for k, v in config.items()},
        "steps": step_log,
        "summary": summary,
    }
    (dest / f"{run_id}.json").write_text(json.dumps(record, indent=2))
    return run_id

A row-count ledger per step is the highest-value diagnostic a GIS pipeline can produce: when the output is smaller than expected, it tells you which step dropped the rows, and when.

Code examples

Example 1: a complete small pipeline

#!/usr/bin/env python3
"""parcels_pipeline.py — extract, clean, enrich, load, record."""
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
import argparse, json, logging, sys, time
import geopandas as gpd
import yaml

log = logging.getLogger("parcels")

# ── steps: pure functions ─────────────────────────────────────────────────
def drop_invalid(gdf):
    keep = gdf.geometry.notna() & ~gdf.geometry.is_empty
    out = gdf.loc[keep].copy()
    out["geometry"] = out.geometry.make_valid()
    return out[out.geometry.is_valid]

def reproject(gdf, crs):
    return gdf if gdf.crs == crs else gdf.to_crs(crs)

def clip_to(gdf, boundary):
    return gpd.clip(gdf, boundary.to_crs(gdf.crs), keep_geom_type=True)

def add_area(gdf, min_area_m2=0):
    out = gdf.copy()
    out["area_m2"] = out.to_crs(out.estimate_utm_crs()).area.round(2)
    return out[out["area_m2"] >= min_area_m2]

# ── the impure shell ──────────────────────────────────────────────────────
def run(config) -> dict:
    started = time.perf_counter()
    gdf = gpd.read_file(config["input"])
    boundary = gpd.read_file(config["boundary"])
    step_log = []

    steps = [
        ("drop_invalid", lambda g: drop_invalid(g)),
        ("reproject", lambda g: reproject(g, config["crs"])),
        ("clip", lambda g: clip_to(g, boundary)),
        ("add_area", lambda g: add_area(g, config["min_area_m2"])),
    ]
    for name, fn in steps:
        t0, before = time.perf_counter(), len(gdf)
        gdf = fn(gdf)
        step_log.append({"step": name, "rows_in": before, "rows_out": len(gdf),
                         "seconds": round(time.perf_counter() - t0, 2)})
        log.info("%-14s %7d → %7d", name, before, len(gdf))

    if gdf.empty:
        raise ValueError("pipeline produced no features")

    dest = Path(config["output"])
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_suffix(dest.suffix + ".part")
    gdf.to_file(tmp, driver="GPKG")
    tmp.replace(dest)

    return {"output": str(dest), "features": len(gdf), "steps": step_log,
            "duration_s": round(time.perf_counter() - started, 2)}

def main() -> int:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)-7s %(message)s")
    ap = argparse.ArgumentParser()
    ap.add_argument("config", type=Path)
    args = ap.parse_args()

    config = yaml.safe_load(args.config.read_text())
    for key in ("input", "boundary", "output"):
        config[key] = (args.config.parent / config[key]).resolve()

    try:
        summary = run(config)
    except Exception:
        log.exception("run failed")
        return 1

    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    Path("logs/runs").mkdir(parents=True, exist_ok=True)
    Path(f"logs/runs/{run_id}.json").write_text(json.dumps(summary, indent=2))
    log.info("done: %s features → %s", summary["features"], summary["output"])
    return 0

if __name__ == "__main__":
    raise SystemExit(main())

Example 2: test a step without any files

import geopandas as gpd
import pytest
from shapely.geometry import Polygon

@pytest.fixture
def squares():
    return gpd.GeoDataFrame(
        {"id": [1, 2]},
        geometry=[Polygon([(0, 0), (100, 0), (100, 100), (0, 100)]),
                  Polygon([(200, 0), (210, 0), (210, 10), (200, 10)])],
        crs="EPSG:27700")

def test_add_area_filters_small_parcels(squares):
    out = add_area(squares, min_area_m2=1000)
    assert len(out) == 1
    assert out["area_m2"].iloc[0] == pytest.approx(10_000, rel=1e-3)

def test_reproject_is_a_no_op_when_crs_matches(squares):
    assert reproject(squares, "EPSG:27700") is squares

Pure steps make the test suite fast enough to run on every save, which is what keeps it being run.

Example 3: a dependency graph, when order is not linear

STAGES = {
    "extract_parcels": {"needs": [], "fn": extract_parcels},
    "extract_owners":  {"needs": [], "fn": extract_owners},
    "clean_parcels":   {"needs": ["extract_parcels"], "fn": clean_parcels},
    "join_owners":     {"needs": ["clean_parcels", "extract_owners"], "fn": join_owners},
    "publish":         {"needs": ["join_owners"], "fn": publish},
}

def topological_order(stages):
    done, order = set(), []
    while len(order) < len(stages):
        ready = [n for n, s in stages.items()
                 if n not in done and all(d in done for d in s["needs"])]
        if not ready:
            raise ValueError("cycle in the pipeline graph")
        for name in sorted(ready):
            order.append(name); done.add(name)
    return order

print(" → ".join(topological_order(STAGES)))

When the graph gets big enough that this matters, that is the signal to adopt a real orchestrator rather than growing your own.

Example 4: the shape of the run record

{
  "run_id": "20260811T023014Z",
  "config": {"input": "data/raw/parcels.gpkg", "crs": "EPSG:27700", "min_area_m2": "5"},
  "steps": [
    {"step": "drop_invalid", "rows_in": 4812, "rows_out": 4801, "seconds": 1.2},
    {"step": "reproject",    "rows_in": 4801, "rows_out": 4801, "seconds": 0.4},
    {"step": "clip",         "rows_in": 4801, "rows_out": 4106, "seconds": 3.9},
    {"step": "add_area",     "rows_in": 4106, "rows_out": 4098, "seconds": 0.6}
  ],
  "summary": {"features": 4098, "output": "data/out/parcels_clean.gpkg", "duration_s": 6.4}
}

Read the ledger and the pipeline explains itself: 695 features were removed by the clip, 8 by the area filter, and nothing else changed the count.

Explanation

The word "pipeline" borrows from Unix pipes, and the analogy is exact: small programs that each do one thing, connected so that one's output is the next one's input. What makes it more than a metaphor in data work is that the connections are typed — each stage declares what it needs and what it produces — and that the whole assembly is described somewhere other than in the order of lines in a script.

Panels contrasting a monolithic script with a structured pipeline across testability, reruns and debugging. — The difference is not size — it is whether the parts can be reasoned about separately.

The reason pure steps matter so much in GIS specifically is that spatial bugs are usually silent. A CRS mistake, a bad join or an over-aggressive filter produces valid output with the wrong values, so the only way to catch it is to test the transformation in isolation with data whose right answer you know. A function that reads a file, transforms it and writes another cannot be tested that way; a function that takes a GeoDataFrame and returns one can be tested with four synthetic squares in milliseconds.

Configuration deserves the same discipline. The moment paths and thresholds live in the code, running last month's data means editing the code, which means the code and the run are no longer separable — and reproducing an old result becomes archaeology. Lifting them into a config file makes the code a constant and the run a variable, which is what makes a result reproducible at all.

Idempotency is the property that lets a pipeline be operated rather than supervised. If re-running is safe, then a failure can be retried, a partial run can be finished, and a suspicious result can be regenerated. If re-running appends rows or accumulates files, every failure requires a human to work out what state the world is in first.

Finally, observability is what stops a pipeline becoming the script nobody dares touch. A per-step row-count ledger and a run record turn "the output looks wrong" into "the clip step dropped 40% instead of 14%, starting on the 14th" — a question with an answer. Without it, the only debugging tool is re-reading 700 lines of code, which is exactly where this article started.

Edge cases or notes

Not everything needs a pipeline: A one-off analysis is fine as a notebook. The structure pays off when something runs repeatedly and unattended.
Steps should be coarse enough to be meaningful: Forty two-line steps is as unreadable as one long function.
Pure does not mean side-effect-free forever: Reading a reference layer inside a step is pragmatic; writing an output from the middle of one is not.
Beware the config that becomes a language: If the YAML has conditionals and loops, the logic has escaped into configuration. Move it back to code.
Airflow is not the beginning: A cron job and an exit code covers most GIS pipelines. Adopt an orchestrator when you genuinely have a dependency graph.
Intermediate outputs are a debugging tool: Writing each stage to disk costs I/O and buys the ability to resume and inspect. Make it a flag.
Schema drift is the usual breakage: Suppliers rename columns. Validate the input schema at the start of every run.

Internal links

FAQ

What makes a script a pipeline?

Structure and separation: discrete steps with declared inputs and outputs, configuration outside the code, deterministic re-runs, and enough logging to explain a result afterwards.

How small should a step be?

Small enough to test on its own and to name in a sentence, large enough to be meaningful in a log line. "Clip to boundary" is a step; "call to_crs" usually is not.

Do I need Airflow or Prefect?

Only when you have a real dependency graph or need backfills, retries and a UI across many jobs. A single linear GIS pipeline is well served by cron plus honest exit codes.

What belongs in configuration?

Paths, CRS, thresholds, date ranges, feature flags — anything that varies between runs without changing what the pipeline does. Logic belongs in code.

Why is idempotency so important?

Because it makes retries safe. A pipeline that appends on every run turns each failure into a manual clean-up before you can try again.

How do I debug a pipeline that produces the wrong numbers?

Read the per-step row counts. They localise the problem to a step immediately, which is why logging rows in and rows out for every step is the highest-value line of code in the whole design.

Should intermediate results be written to disk?

For long pipelines, yes — behind a flag. It costs I/O and buys resumability and the ability to inspect what a step actually produced.

What Is a GIS Data Pipeline? The Anatomy Explained #

Problem statement #

Quick answer #

The anatomy #

Step-by-step solution #

Steps: one job, one function, no I/O #

Contracts: what each step promises #

Configuration: everything that changes without the logic changing #

Composition: chaining steps so the pipeline is data #

Idempotency: safe to run twice #

Orchestration: what runs it, and what happens when it fails #

Observability: the part that makes it maintainable #

Code examples #

Example 1: a complete small pipeline #

Example 2: test a step without any files #

Example 3: a dependency graph, when order is not linear #

Example 4: the shape of the run record #

Explanation #

Edge cases or notes #

Internal links #

FAQ #

What makes a script a pipeline? #

How small should a step be? #

Do I need Airflow or Prefect? #

What belongs in configuration? #

Why is idempotency so important? #

How do I debug a pipeline that produces the wrong numbers? #

Should intermediate results be written to disk? #