How to Build a GIS Data Pipeline in Python: The Complete Workflow

There is a moment in every GIS project when the notebook stops being enough. The analysis works, the map looks right, and then someone asks for it again next month with fresh data — and you discover that "the analysis" is forty cells you have to run in the right order while remembering which three you must skip. A pipeline is the cure. It is not a framework you install; it is a shape you give your own code so that one command takes raw input to finished output, tells you what it did, and does the same thing next month. This article builds that shape from scratch, stage by stage.

Problem statement

You have working GIS code, but running it is a manual ritual. You are hitting some mix of these:

The order only exists in your head — cells or scripts have to run in a particular sequence and nothing enforces it.
Paths are hard-coded everywhere — moving to a new folder or a new month of data means find-and-replace across the file.
Rerunning redoes everything — a change to the last step forces a rerun of the slow reprojection at the start.
Failures are silent or fatal — a bad input either crashes the whole thing or, worse, quietly produces an empty layer nobody notices.
You cannot prove what happened — six weeks later there is no record of which inputs produced which output.

The goal: one script you can run with a single command that reads from a defined input location, moves data through named stages, writes versioned output, checks its own work, and leaves behind a log and a manifest of exactly what it did.

Quick answer

Model the workflow as a list of small, named stages. Each stage is a plain Python function that takes the previous stage's output and returns the next one. A tiny runner loops over the stages, times them, logs them, and stops on the first real failure. Paths and parameters live in one config dictionary at the top, never inside the functions.

A five-stage GIS pipeline: extract, clean, transform, load, and report, each stage feeding the next. — Every pipeline is the same five moves — the GIS work lives inside the stages, not in the runner.

import logging
from pathlib import Path
import geopandas as gpd

CONFIG = {
    "source": Path("data/raw/parcels.gpkg"),
    "output_dir": Path("data/processed"),
    "target_crs": "EPSG:27700",
    "min_area_m2": 50,
}

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("pipeline")

def extract(cfg):
    return gpd.read_file(cfg["source"])

def clean(gdf, cfg):
    gdf = gdf[gdf.geometry.notna() & ~gdf.geometry.is_empty]
    gdf["geometry"] = gdf.geometry.make_valid()
    return gdf

def transform(gdf, cfg):
    gdf = gdf.to_crs(cfg["target_crs"])
    gdf["area_m2"] = gdf.geometry.area
    return gdf[gdf["area_m2"] >= cfg["min_area_m2"]]

def load(gdf, cfg):
    cfg["output_dir"].mkdir(parents=True, exist_ok=True)
    out = cfg["output_dir"] / "parcels_clean.gpkg"
    gdf.to_file(out, driver="GPKG")
    return out

STAGES = [extract, clean, transform, load]

def run(cfg):
    data = None
    for stage in STAGES:
        log.info("stage: %s", stage.__name__)
        data = stage(cfg) if data is None else stage(data, cfg)
    log.info("done → %s", data)
    return data

if __name__ == "__main__":
    run(CONFIG)

That is a complete pipeline in fifty lines. The rest of this article explains why each part is shaped that way, and how to grow it without turning it back into a mess.

Step-by-step solution

Separate the runner from the work

The single most useful decision is to keep the sequencing code away from the GIS code. Stage functions know about GeoDataFrames and CRSs; the runner knows about order, timing, and logging, and never touches geometry. Once that line is drawn, you can add retries, caching, or a progress bar to the runner without editing a single line of your analysis.

def run(cfg, stages):
    """The runner. It knows nothing about GIS — only about order."""
    data = None
    for stage in stages:
        data = stage(cfg) if data is None else stage(data, cfg)
    return data

Put every path and parameter in one config

Hard-coded paths are what make a script un-runnable by anyone else, including you on a different machine. Collect them at the top in a single dictionary. Later you can move that dictionary into a YAML config file or expose it as command-line flags without touching the stages.

CONFIG = {
    "source": Path("data/raw/parcels.gpkg"),
    "boundary": Path("data/raw/district.geojson"),
    "output_dir": Path("data/processed"),
    "target_crs": "EPSG:27700",
    "min_area_m2": 50,
}

The rule that keeps this honest: no stage function may contain a literal path. If a function needs a file, it takes it from cfg.

Make the extract stage boring and strict

The extract stage does one thing — get data in — and it should complain immediately if what arrived is not what you expected. Failing loudly at the first stage is far cheaper than failing subtly at the last.

def extract(cfg):
    src = cfg["source"]
    if not src.exists():
        raise FileNotFoundError(f"source missing: {src}")
    gdf = gpd.read_file(src)
    if gdf.empty:
        raise ValueError(f"source has no features: {src}")
    if gdf.crs is None:
        raise ValueError(f"source has no CRS: {src} — set it before running")
    log.info("read %d features from %s (%s)", len(gdf), src.name, gdf.crs)
    return gdf

Give the clean stage its own home

Nearly every real pipeline needs a cleaning stage, and it is always the stage that grows. Keeping it separate means you can develop it against a single file and reuse it across pipelines. The specifics — invalid geometries, nulls, duplicates, CRS repair — are covered in depth in the Python GIS data cleaning checklist.

def clean(gdf, cfg):
    before = len(gdf)
    gdf = gdf[gdf.geometry.notna() & ~gdf.geometry.is_empty].copy()
    invalid = ~gdf.geometry.is_valid
    if invalid.any():
        log.warning("repairing %d invalid geometries", int(invalid.sum()))
        gdf.loc[invalid, "geometry"] = gdf.loc[invalid, "geometry"].make_valid()
    log.info("clean: %d → %d features", before, len(gdf))
    return gdf

Keep the transform stage pure

The transform stage is your actual analysis: reproject, clip, join, aggregate. Write it so it takes a GeoDataFrame and returns a GeoDataFrame and touches nothing on disk. Pure functions are the ones you can test, and testing the analysis in isolation is what stops a pipeline from silently drifting.

def transform(gdf, cfg):
    gdf = gdf.to_crs(cfg["target_crs"])          # metric CRS before any measurement
    gdf["area_m2"] = gdf.geometry.area
    keep = gdf["area_m2"] >= cfg["min_area_m2"]
    log.info("transform: dropping %d parcels under %s m²",
             int((~keep).sum()), cfg["min_area_m2"])
    return gdf.loc[keep].copy()

Write output where you can find it again

The load stage writes results — and it should never overwrite the previous run without being asked. Writing into a dated run folder means yesterday's output is still there when today's looks wrong.

from datetime import date

def load(gdf, cfg):
    run_dir = cfg["output_dir"] / date.today().isoformat()
    run_dir.mkdir(parents=True, exist_ok=True)
    out = run_dir / "parcels_clean.gpkg"
    gdf.to_file(out, driver="GPKG")
    log.info("wrote %d features → %s", len(gdf), out)
    return out

Finish with a report stage

The last stage produces the artefact that tells a human what happened: feature counts in and out, CRS, runtime, output path. This is what turns "it ran" into "it ran and here is the evidence".

A dated run folder containing the config snapshot, the log file, the output data, and a run manifest. — One folder per run: the config that produced it, the log, the data, and a manifest.

import json, time

def report(out_path, cfg, stats):
    manifest = {
        "output": str(out_path),
        "source": str(cfg["source"]),
        "target_crs": cfg["target_crs"],
        "features_in": stats["features_in"],
        "features_out": stats["features_out"],
        "seconds": round(stats["seconds"], 1),
    }
    (out_path.parent / "run_manifest.json").write_text(json.dumps(manifest, indent=2))
    return manifest

Code examples

Example 1: The smallest pipeline that deserves the name

Three functions and a loop. If your workflow is simple, stop here — this is already reproducible and readable.

from pathlib import Path
import geopandas as gpd

CONFIG = {"source": Path("data/raw/roads.shp"),
          "out": Path("data/processed/roads.gpkg"),
          "crs": "EPSG:3035"}

def extract(cfg):
    return gpd.read_file(cfg["source"])

def transform(gdf, cfg):
    gdf = gdf.to_crs(cfg["crs"])
    gdf["length_m"] = gdf.geometry.length
    return gdf

def load(gdf, cfg):
    cfg["out"].parent.mkdir(parents=True, exist_ok=True)
    gdf.to_file(cfg["out"], driver="GPKG")
    return cfg["out"]

data = extract(CONFIG)
data = transform(data, CONFIG)
print("wrote", load(data, CONFIG))

Example 2: A runner that times and logs every stage

Once the runner owns sequencing, adding instrumentation costs three lines and benefits every stage at once.

import time, logging

log = logging.getLogger("pipeline")

def run(cfg, stages):
    data, timings = None, {}
    for stage in stages:
        t0 = time.perf_counter()
        data = stage(cfg) if data is None else stage(data, cfg)
        timings[stage.__name__] = time.perf_counter() - t0
        log.info("%-10s %6.2fs", stage.__name__, timings[stage.__name__])
    slowest = max(timings, key=timings.get)
    log.info("slowest stage: %s (%.2fs)", slowest, timings[slowest])
    return data, timings

Example 3: Stop on failure, but say exactly where

A pipeline that dies with a bare traceback wastes your time. Wrap the stage call so the error message names the stage that failed and preserves the original exception.

class StageError(RuntimeError):
    pass

def run(cfg, stages):
    data = None
    for stage in stages:
        try:
            data = stage(cfg) if data is None else stage(data, cfg)
        except Exception as exc:
            raise StageError(f"stage '{stage.__name__}' failed: {exc}") from exc
    return data

raise … from exc keeps the original traceback attached, so you still see the GDAL error underneath the friendly message.

Example 4: A pipeline with two inputs

Real workflows rarely have one source. Pass a dictionary between stages instead of a bare GeoDataFrame as soon as you need more than one layer in flight.

def extract(cfg):
    return {
        "parcels": gpd.read_file(cfg["source"]),
        "boundary": gpd.read_file(cfg["boundary"]),
    }

def clip_to_boundary(state, cfg):
    parcels, boundary = state["parcels"], state["boundary"]
    boundary = boundary.to_crs(parcels.crs)          # always align CRS first
    state["parcels"] = gpd.clip(parcels, boundary)
    return state

def load(state, cfg):
    out = cfg["output_dir"] / "parcels_in_district.gpkg"
    state["parcels"].to_file(out, driver="GPKG")
    return out

Example 5: The full script, end to end

Config, stages, runner, manifest — the shape you can copy into a new project and fill in.

import json, logging, time
from datetime import datetime
from pathlib import Path
import geopandas as gpd

CONFIG = {
    "source": Path("data/raw/parcels.gpkg"),
    "output_dir": Path("data/processed"),
    "target_crs": "EPSG:27700",
    "min_area_m2": 50,
}

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s %(levelname)-7s %(message)s")
log = logging.getLogger("pipeline")

def extract(cfg):
    gdf = gpd.read_file(cfg["source"])
    if gdf.crs is None:
        raise ValueError("source has no CRS")
    return gdf

def clean(gdf, cfg):
    gdf = gdf[gdf.geometry.notna() & ~gdf.geometry.is_empty].copy()
    gdf["geometry"] = gdf.geometry.make_valid()
    return gdf

def transform(gdf, cfg):
    gdf = gdf.to_crs(cfg["target_crs"])
    gdf["area_m2"] = gdf.geometry.area
    return gdf[gdf["area_m2"] >= cfg["min_area_m2"]].copy()

def load(gdf, cfg):
    run_dir = cfg["output_dir"] / datetime.now().strftime("%Y-%m-%d")
    run_dir.mkdir(parents=True, exist_ok=True)
    out = run_dir / "parcels_clean.gpkg"
    gdf.to_file(out, driver="GPKG")
    return out

def main(cfg):
    started = time.perf_counter()
    features_in = None
    data = None
    for stage in (extract, clean, transform, load):
        t0 = time.perf_counter()
        data = stage(cfg) if data is None else stage(data, cfg)
        if stage is extract:
            features_in = len(data)
        log.info("%-10s %6.2fs", stage.__name__, time.perf_counter() - t0)

    manifest = {
        "run_at": datetime.now().isoformat(timespec="seconds"),
        "source": str(cfg["source"]),
        "output": str(data),
        "features_in": features_in,
        "target_crs": cfg["target_crs"],
        "seconds": round(time.perf_counter() - started, 1),
    }
    (Path(data).parent / "run_manifest.json").write_text(json.dumps(manifest, indent=2))
    log.info("pipeline finished in %.1fs", manifest["seconds"])
    return manifest

if __name__ == "__main__":
    main(CONFIG)

Explanation

A pipeline earns its keep for one reason: it separates what the workflow does from how it is run. The stage list is a readable description of the workflow — anyone can see that data is extracted, cleaned, transformed, and loaded, in that order, without reading a single line of GeoPandas. The runner is a piece of machinery you write once and reuse forever.

That separation is what makes every later improvement cheap. Timing every stage is a change to the runner. So is stopping on failure with a useful message, retrying a flaky network read, skipping stages whose inputs have not changed, and sending an alert when a scheduled run dies. None of those touch your GIS code, which is precisely why they are safe to add to a workflow that is already producing numbers people rely on.

One long script versus a staged pipeline: tangled and unrerunnable on the left, named and instrumented stages on the right. — Same GIS code, different shape — the pipeline version can be logged, tested, and rerun.

Passing data between stages in memory, rather than through files, is the right default. It is faster, it avoids a folder full of step2_tmp.gpkg, and it keeps each stage honest about its inputs. Write to disk only where the artefact matters: the final output, and any intermediate a human will actually open. The exception is a long or expensive stage — persisting its result is what makes caching possible later.

The stages themselves should stay small enough to describe in a sentence. When a transform function starts needing its own section headings, split it. Two stages named join_census and aggregate_by_ward tell you more from the log than one stage called process, and when the run fails at 3 a.m., the stage name in the log is the first thing you read.

Edge cases or notes

Stages that need more than one input

The single-value chain (data = stage(data, cfg)) is the simplest thing that works, and it breaks the moment a stage needs two layers. Pass a dictionary of named layers instead — a "state" object — as shown in Example 4. Keep the keys stable and log them, so you can see what is in flight at each stage.

Big data that will not fit in memory

Passing GeoDataFrames between stages assumes the dataset fits in RAM. When it does not, switch the contract: each stage takes and returns a path rather than a frame, and the data lives on disk between stages. It is slower but bounded. For a folder of many moderate files, the better answer is usually a batch loop inside a single stage.

Partial failure across many inputs

A pipeline over one dataset should stop on failure. A pipeline over two hundred files should not — one corrupt file must not sink the run. That is a batch job, and the right pattern is a per-file try/except with a summary report; see batch process a folder of GIS files and log and summarise errors.

Reprojection order

Reproject once, early, and measure afterwards. Calling to_crs() inside three different stages is both slow and a source of subtle disagreement between outputs. If stages need different CRSs, make that explicit with a stage named for it rather than scattering conversions.

Nothing came out, but nothing failed

The classic silent pipeline failure: a clip against a mismatched CRS returns zero features, and the run "succeeds" with an empty file. Guard every stage that can empty a frame with an explicit check, or add a validation stage — see validate pipeline inputs and outputs.

When to reach for a real orchestrator

Airflow, Prefect, and Dagster are excellent when you have many pipelines, real dependencies between them, a team, and infrastructure to run them on. For a single scheduled workflow on one machine, they add more operational surface than they remove. Start with the script in this article and a scheduled run; move up when you genuinely have a graph rather than a line.

Internal links

FAQ

What is a GIS data pipeline?

A pipeline is a workflow expressed as an ordered list of small, named stages, run by one command. Each stage does a single job — extract, clean, transform, load, report — and hands its result to the next. It is a shape you give ordinary Python code, not a library you install, and it exists so a workflow can be rerun months later without you remembering anything.

Do I need Airflow or Prefect to build a pipeline?

No, and for a single workflow on one machine you probably should not. Those tools solve orchestration across many pipelines with real infrastructure behind them. A list of functions, a runner loop, a config dictionary, and a scheduled run cover the great majority of GIS automation with a fraction of the moving parts. Move up when you have a genuine dependency graph and a team.

Should stages pass GeoDataFrames or file paths?

Pass GeoDataFrames in memory by default — it is faster and avoids a litter of temporary files. Switch to paths when the data no longer fits in RAM, or when an intermediate result is expensive enough that you want it cached on disk between runs. You can mix: most stages pass frames, and one slow stage persists its output.

How do I stop the pipeline from overwriting yesterday's results?

Write into a dated run folder — data/processed/2026-08-01/ — created by the load stage. Yesterday's output stays untouched, comparing two runs is a diff of two folders, and a bad run never destroys a good one. Add a run_manifest.json beside the data so each folder explains itself.

Where should logging go?

Configure logging once in the entry point, never inside stage functions. Stages call log.info() and stay ignorant of where that goes. Then a single change sends the same messages to the console interactively and to a file when scheduled. The pattern is covered in detail in log and summarise errors in a batch GIS job.

How do I test a pipeline?

Test the stage functions, not the runner. Because each transform stage is a pure function from GeoDataFrame to GeoDataFrame, you can build a three-row GeoDataFrame in the test, run the stage, and assert on the result — no files, no fixtures, milliseconds per test. Then add one end-to-end smoke run against a tiny sample dataset to catch wiring mistakes.

What belongs in the config and what belongs in the code?

Anything that changes between runs or between machines goes in the config: paths, CRSs, thresholds, dates, credentials. Anything that defines what the workflow is stays in code: the stage list and the logic inside each stage. If you find yourself adding a config flag that switches on a different algorithm, that is usually a sign you have two pipelines, not one.

How do I know the run actually produced good output?

Do not rely on the absence of an exception. Add explicit assertions at stage boundaries — feature count above zero, CRS as expected, no null geometries — and write a manifest recording counts in and out. A pipeline that finishes with an empty output has failed, and only a check will tell you so; see validate pipeline inputs and outputs.

How to Build a GIS Data Pipeline in Python: The Complete Workflow #

Problem statement #

Quick answer #

Step-by-step solution #

Separate the runner from the work #

Put every path and parameter in one config #

Make the extract stage boring and strict #

Give the clean stage its own home #

Keep the transform stage pure #

Write output where you can find it again #

Finish with a report stage #

Code examples #

Example 1: The smallest pipeline that deserves the name #

Example 2: A runner that times and logs every stage #

Example 3: Stop on failure, but say exactly where #

Example 4: A pipeline with two inputs #

Example 5: The full script, end to end #

Explanation #

Edge cases or notes #

Stages that need more than one input #

Big data that will not fit in memory #

Partial failure across many inputs #

Reprojection order #

Nothing came out, but nothing failed #

When to reach for a real orchestrator #

Internal links #

FAQ #

What is a GIS data pipeline? #

Do I need Airflow or Prefect to build a pipeline? #

Should stages pass GeoDataFrames or file paths? #

How do I stop the pipeline from overwriting yesterday's results? #

Where should logging go? #

How do I test a pipeline? #

What belongs in the config and what belongs in the code? #

How do I know the run actually produced good output? #