How to Record Run Metadata and Data Lineage in a GIS Pipeline

Problem statement

Someone opens a map and asks a reasonable question:

"This layer says 4,812 parcels. The council's website says 4,790. Which is right, and when was ours made?"

You look at the file. It is called parcels_final.gpkg, modified three weeks ago. You do not know which input extract it came from, which version of the code produced it, what CRS the source was in, whether the invalid-geometry repair was enabled that day, or who ran it.

That is a lineage failure, and it is expensive out of proportion to its cause. The fix is small: every run writes a record of what it consumed, what it did, and what it produced — automatically, as part of the run.

What you need to be able to answer later:

• which input files, with what checksums and modification dates?
• which version of the code and which library versions?
• what parameters — CRS, thresholds, flags?
• what came out: counts, extents, checksums?
• when, on what machine, by whom, and how long did it take?
• which steps dropped rows, and how many?

Quick answer

Write a JSON run record next to every output, and stamp the output itself:

1. capture inputs — path, size, mtime, checksum — before processing
2. capture the environment — git revision, Python and key library versions
3. capture parameters — the resolved config, not the defaults you assume
4. capture outputs — counts, bounds, CRS, checksum
5. write it as JSON per run, and append a one-line summary to a ledger

import hashlib, json, platform, subprocess, sys, time
from datetime import datetime, timezone
from pathlib import Path
import geopandas as gpd

def sha256(path: Path, chunk=1 << 20) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        for block in iter(lambda: fh.read(chunk), b""):
            h.update(block)
    return h.hexdigest()

def git_rev() -> str | None:
    try:
        return subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True,
                              text=True, check=True).stdout.strip()
    except Exception:
        return None

src = Path("data/raw/parcels.gpkg")
dest = Path("data/out/parcels_clean.gpkg")
started = time.perf_counter()

gdf = gpd.read_file(src)
clean = gdf[gdf.geometry.is_valid].to_crs("EPSG:27700")
dest.parent.mkdir(parents=True, exist_ok=True)
clean.to_file(dest, driver="GPKG")

record = {
    "run_id": datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"),
    "started_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"),
    "duration_s": round(time.perf_counter() - started, 2),
    "environment": {"git_rev": git_rev(), "python": sys.version.split()[0],
                    "geopandas": gpd.__version__},
    "host": {"hostname": platform.node(), "platform": platform.platform()},
    "inputs": [{"path": str(src), "sha256": sha256(src),
                "features": len(gdf), "crs": str(gdf.crs)}],
    "parameters": {"target_crs": "EPSG:27700", "drop_invalid": True},
    "outputs": [{"path": str(dest), "sha256": sha256(dest),
                 "features": len(clean), "crs": str(clean.crs),
                 "bounds": [round(v, 3) for v in clean.total_bounds]}],
    "counts": {"in": len(gdf), "out": len(clean), "dropped": len(gdf) - len(clean)},
    "status": "ok",
}
Path(f"logs/runs/{record['run_id']}.json").parent.mkdir(parents=True, exist_ok=True)
Path(f"logs/runs/{record['run_id']}.json").write_text(json.dumps(record, indent=2))
print(json.dumps(record["counts"]), "| input sha256", record["inputs"][0]["sha256"][:16], "…")

{"in": 21253, "out": 21253, "dropped": 0} | input sha256 1580163716b05c9d …

Run on the Land Registry's Rutland parcels (set up in The data used on this page), nothing was dropped — that file has no invalid polygons — and the record says so rather than leaving it to be assumed. The input checksum is the piece that turns a vague record into evidence: it identifies the exact bytes the run consumed, whatever the file was called.

What a run record holds

Run ids, inputs, environment, parameters and steps

Give every run an id

from datetime import datetime, timezone
import uuid

def new_run_id() -> str:
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    return f"{stamp}-{uuid.uuid4().hex[:6]}"

RUN_ID = new_run_id()
print(RUN_ID, new_run_id())

20261010T223851Z-115d10 20261010T223851Z-c974f1

A timestamp sorts naturally; the random suffix keeps two runs in the same second distinct. Put the id in the log lines, the output metadata and the filenames, and every artefact of a run becomes findable from any other.

Record the inputs before you touch them

from pathlib import Path
from datetime import datetime, timezone
import pyogrio

def describe_input(path: Path) -> dict:
    stat = path.stat()
    info = pyogrio.read_info(path)
    return {
        "path": str(path.resolve()),
        "bytes": stat.st_size,
        "modified_utc": datetime.fromtimestamp(stat.st_mtime, timezone.utc)
                        .isoformat(timespec="seconds"),
        "sha256": sha256(path),
        "features": int(info["features"]),
        "crs": info["crs"],
        "geometry_type": info["geometry_type"],
        "fields": list(info["fields"]),
    }


described = describe_input(src)
print({k: (v if k != "path" else Path(v).name) for k, v in described.items() if k != "sha256"})

{'path': 'parcels.gpkg', 'bytes': 20963328, 'modified_utc': '2026-10-10T22:38:51+00:00', 'features': 21253, 'crs': 'EPSG:27700', 'geometry_type': 'Polygon', 'fields': ['gml_id', 'INSPIREID', 'LABEL', 'NATIONALCADASTRALREFERENCE', 'VALIDFROM', 'BEGINLIFESPANVERSION']}

Hashing a multi-gigabyte input on every run is wasteful; a size-plus-mtime fingerprint is usually enough, with a full hash reserved for deliverables:

def quick_fingerprint(path: Path) -> str:
    stat = path.stat()
    return f"{stat.st_size}-{int(stat.st_mtime)}"


start = time.perf_counter(); full = sha256(src); full_s = time.perf_counter() - start
start = time.perf_counter(); quick = quick_fingerprint(src); quick_s = time.perf_counter() - start
print(f"sha256 of {src.stat().st_size / 1e6:.1f} MB: {full_s * 1000:.0f} ms | fingerprint {quick}: {quick_s * 1e6:.0f} µs")

sha256 of 21.0 MB: 18 ms | fingerprint 20963328-1791671931: 41 µs

Record the environment that produced the result

import os
import platform
import subprocess
import sys


def describe_environment() -> dict:
    def run(cmd):
        try:
            return subprocess.run(cmd, capture_output=True, text=True, check=True).stdout.strip()
        except Exception:
            return None

    import geopandas, shapely, pyproj, rasterio
    return {
        "git_rev": run(["git", "rev-parse", "HEAD"]),
        "git_dirty": bool(run(["git", "status", "--porcelain", "--untracked-files=no"])),
        "python": sys.version.split()[0],
        "executable": sys.executable,
        "packages": {
            "geopandas": geopandas.__version__,
            "shapely": shapely.__version__,
            "pyproj": pyproj.__version__,
            "rasterio": rasterio.__version__,
            "gdal": rasterio.__gdal_version__,
            "proj": pyproj.proj_version_str,
        },
        "hostname": platform.node(),
        "platform": platform.platform(),
        "container_image": os.environ.get("IMAGE_TAG"),
    }


environment = describe_environment()
print({k: environment[k] for k in ["git_rev", "git_dirty", "python"]}, environment["packages"])

{'git_rev': '6bfec3f2daaa84bdda90023257878f2b84136775', 'git_dirty': False, 'python': '3.14.4'} {'geopandas': '1.1.4', 'shapely': '2.1.2', 'pyproj': '3.8.0', 'rasterio': '1.5.1', 'gdal': '3.12.4', 'proj': '9.8.1'}

git_dirty is worth its two lines: a result produced from uncommitted changes is not reproducible, and knowing that six months later saves a long argument. Pass --untracked-files=no: plain git status --porcelain also counts untracked files, and here the working folder's scratch files alone made it report the committed code as dirty. GDAL and PROJ versions belong here too — a coordinate-operation change between PROJ releases can shift results by metres.

Record the parameters that were actually used

def describe_parameters(config: dict, cli_args) -> dict:
    return {
        "config_file": str(config.get("_source_path")),
        "config_sha256": sha256(Path(config["_source_path"])) if config.get("_source_path") else None,
        "resolved": {k: str(v) for k, v in sorted(config.items()) if not k.startswith("_")},
        "cli_overrides": {k: v for k, v in vars(cli_args).items() if v is not None},
        "env_overrides": {k: v for k, v in os.environ.items() if k.startswith("GIS_")},
    }


import argparse

config = {**json.loads(Path("config/parcels.json").read_text()), "_source_path": "config/parcels.json"}
args = argparse.Namespace(target_crs="EPSG:27700", dry_run=None)
os.environ["GIS_MIN_AREA_M2"] = "50"
print(json.dumps(describe_parameters(config, args), indent=1)[:400])

{
 "config_file": "config/parcels.json",
 "config_sha256": "b1a5a332d6cb159e9525e6c1b568ed0aed11a9aea25ff6608254a85b24b3dcc4",
 "resolved": {
  "drop_invalid": "True",
  "source": "data/raw/parcels.gpkg",
  "target_crs": "EPSG:27700"
 },
 "cli_overrides": {
  "target_crs": "EPSG:27700"
 },
 "env_overrides": {
  "GIS_MIN_AREA_M2": "50"
 }
}

Record the resolved values — after defaults, config file, environment variables and command-line flags have all been merged. The config file alone does not tell you what ran.

Record each step, not just the run

Per-step counts are what let you answer "where did those 300 rows go?".

import time
from contextlib import contextmanager

class Lineage:
    def __init__(self, run_id: str):
        self.run_id, self.steps = run_id, []

    @contextmanager
    def step(self, name: str, **params):
        started = time.perf_counter()
        entry = {"name": name, "params": params, "started": time.time()}
        try:
            yield entry
            entry["status"] = "ok"
        except Exception as exc:
            entry["status"] = "failed"
            entry["error"] = f"{type(exc).__name__}: {exc}"
            raise
        finally:
            entry["duration_s"] = round(time.perf_counter() - started, 3)
            self.steps.append(entry)

lineage = Lineage(RUN_ID)

with lineage.step("read", path=str(src)) as s:
    gdf = gpd.read_file(src)
    s["rows_out"] = len(gdf)

with lineage.step("drop_invalid") as s:
    s["rows_in"] = len(gdf)
    gdf = gdf[gdf.geometry.is_valid]
    s["rows_out"] = len(gdf)
    s["rows_dropped"] = s["rows_in"] - s["rows_out"]

with lineage.step("reproject", target="EPSG:27700") as s:
    s["crs_in"] = str(gdf.crs)
    gdf = gdf.to_crs("EPSG:27700")
    s["crs_out"] = str(gdf.crs)

for entry in lineage.steps:
    print({k: v for k, v in entry.items() if k not in ("started",)})

{'name': 'read', 'params': {'path': 'data/raw/parcels.gpkg'}, 'rows_out': 21253, 'status': 'ok', 'duration_s': 0.142}
{'name': 'drop_invalid', 'params': {}, 'rows_in': 21253, 'rows_out': 21253, 'rows_dropped': 0, 'status': 'ok', 'duration_s': 0.052}
{'name': 'reproject', 'params': {'target': 'EPSG:27700'}, 'crs_in': 'EPSG:27700', 'crs_out': 'EPSG:27700', 'status': 'ok', 'duration_s': 0.002}

A row-count ledger through the whole pipeline turns "the output is smaller than I expected" into a specific step and a specific number.

Stamp the output itself

A record beside the file can be separated from it. Put the essentials inside the data too.

import json, sqlite3
from pathlib import Path

def stamp_geopackage(path: Path, record: dict) -> None:
    """Store run metadata in the GeoPackage's own metadata tables."""
    with sqlite3.connect(path) as conn:
        conn.execute("""CREATE TABLE IF NOT EXISTS gpkg_metadata (
            id INTEGER PRIMARY KEY, md_scope TEXT NOT NULL DEFAULT 'dataset',
            md_standard_uri TEXT NOT NULL, mime_type TEXT NOT NULL DEFAULT 'text/xml',
            metadata TEXT NOT NULL)""")
        conn.execute(
            "INSERT INTO gpkg_metadata (md_scope, md_standard_uri, mime_type, metadata) "
            "VALUES ('dataset', 'https://www.spatialworkflow.io/lineage', 'application/json', ?)",
            (json.dumps(record),))

# simplest portable alternative: a sidecar next to the output
def write_sidecar(output: Path, record: dict) -> Path:
    side = output.with_suffix(output.suffix + ".run.json")
    side.write_text(json.dumps(record, indent=2), encoding="utf-8")
    return side


stamp_geopackage(dest, record)
with sqlite3.connect(dest) as conn:
    stored = json.loads(conn.execute("SELECT metadata FROM gpkg_metadata ORDER BY id DESC LIMIT 1").fetchone()[0])
print("stamped run:", stored["run_id"], "| GDAL still reads it:", len(gpd.read_file(dest)), "features",
      "| GDAL sees the metadata:", pyogrio.read_info(dest)["dataset_metadata"])
print("sidecar:", write_sidecar(dest, record).name)

stamped run: 20261010T223851Z | GDAL still reads it: 21253 features | GDAL sees the metadata: None
sidecar: parcels_clean.gpkg.run.json

The hand-made row is there — but GDAL does not see it: GDAL reads gpkg_metadata rows only in its own format. Worse, it is not harmless. Writing the same layer again with GDAL's own metadata= option, to the stamped file and to a fresh one:

fresh = dest.with_name("parcels_clean_fresh.gpkg")
for target in (dest, fresh):
    clean.to_file(target, driver="GPKG", metadata={"run_id": record["run_id"],
                                                   "source_sha256": record["inputs"][0]["sha256"][:16]})
    print(f"{target.name:28} layer metadata read back: {pyogrio.read_info(target)['layer_metadata']}")

parcels_clean.gpkg           layer metadata read back: None
parcels_clean_fresh.gpkg     layer metadata read back: {'run_id': '20261010T223851Z', 'source_sha256': '1580163716b05c9d'}

On the fresh file GDAL stores the metadata and reads it back; on the file with the hand-made table it silently stores nothing — no error, no warning. Whatever GDAL makes of a metadata table it did not create, the effect is that its own metadata for that file is lost. So use GDAL's route rather than sqlite3: metadata= is stored as layer metadata in GDAL's own entry, where any GDAL-based reader finds it. (With pyogrio.write_dataframe() you can set dataset_metadata= and layer_metadata= separately.) The sidecar and the columns below need no cooperation from any driver.

Adding columns works too, and travels with the rows through any format conversion — though a shapefile would cut produced_utc to its ten-character limit:

gdf["run_id"] = RUN_ID
gdf["produced_utc"] = datetime.now(timezone.utc).isoformat(timespec="seconds")

Keep a ledger you can query

import json
from pathlib import Path

LEDGER = Path("logs/runs.jsonl")

def append_ledger(record: dict) -> None:
    LEDGER.parent.mkdir(parents=True, exist_ok=True)
    summary = {
        "run_id": record["run_id"],
        "started_utc": record["started_utc"],
        "duration_s": record["duration_s"],
        "status": record.get("status", "ok"),
        "git_rev": record["environment"]["git_rev"],
        "in_features": record["counts"]["in"],
        "out_features": record["counts"]["out"],
        "output": record["outputs"][0]["path"] if record["outputs"] else None,
    }
    with open(LEDGER, "a", encoding="utf-8") as fh:
        fh.write(json.dumps(summary) + "\n")

import pandas as pd

append_ledger(record)
runs = pd.read_json("logs/runs.jsonl", lines=True)
print(runs.tail(10)[["run_id", "duration_s", "out_features", "status"]].to_string(index=False))
print("\nfeature-count changes:")
print(runs["out_features"].diff().tail(10).to_string())

          run_id  duration_s  out_features status
20261010T223851Z        0.51         21253     ok

feature-count changes:
0   NaN

JSON Lines appends cheaply and reads as a DataFrame, which is all the infrastructure most pipelines ever need.

Tracing Rutland outputs back to their sources

The data used on this page

The input is HM Land Registry's INSPIRE Index Polygons for Rutland — 21,253 freehold parcels in British National Grid — downloaded as GML and saved as data/raw/parcels.gpkg. The setup also puts the working folder under git with one commit, so the records have a revision to report, and writes the small config file that describe_parameters() reads.

import io
import json
import sqlite3
import subprocess
import time
import zipfile
from pathlib import Path

import geopandas as gpd
import pyogrio
import requests

pyogrio.set_gdal_config_options({"GML_DOWNLOAD_SCHEMA": "NO"})
Path("data/raw").mkdir(parents=True, exist_ok=True)
if not Path("data/raw/parcels.gpkg").exists():
    reply = requests.Session().get("https://use-land-property-data.service.gov.uk/datasets/inspire/download/"
                                   "Rutland.zip", timeout=600)
    reply.raise_for_status()
    zipfile.ZipFile(io.BytesIO(reply.content)).extract("Land_Registry_Cadastral_Parcels.gml", "downloads")
    gpd.read_file("downloads/Land_Registry_Cadastral_Parcels.gml").to_file("data/raw/parcels.gpkg")
Path("config").mkdir(exist_ok=True)
Path("config/parcels.json").write_text(json.dumps({"source": "data/raw/parcels.gpkg", "target_crs": "EPSG:27700",
                                                   "drop_invalid": True}))
if not Path(".git").exists():
    Path(".gitignore").write_text("data/\nlogs/\ndownloads/\n")
    for cmd in (["git", "init", "-q"], ["git", "add", ".gitignore", "config"],
                ["git", "-c", "user.name=pipeline", "-c", "user.email=pipeline@example.org",
                 "commit", "-qm", "pipeline config"]):
        subprocess.run(cmd, check=True)
print(pyogrio.read_info("data/raw/parcels.gpkg")["features"], "parcels;",
      "git", subprocess.run(["git", "rev-parse", "--short", "HEAD"], capture_output=True, text=True).stdout.strip())

21253 parcels; git 6bfec3f

Example 1: a complete lineage recorder

%%writefile lineage.py
"""lineage.py — record what a run consumed, did, and produced."""
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
import hashlib, json, os, platform, subprocess, sys, time, uuid

class RunRecorder:
    def __init__(self, name: str, out_dir: Path = Path("logs/runs")):
        self.name = name
        self.run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + uuid.uuid4().hex[:6]
        self.out_dir = out_dir
        self.started = time.perf_counter()
        self.record = {
            "run_id": self.run_id, "pipeline": name,
            "started_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"),
            "environment": self._environment(), "parameters": {},
            "inputs": [], "outputs": [], "steps": [], "status": "running",
        }

    @staticmethod
    def _sha256(path: Path, chunk=1 << 20) -> str:
        h = hashlib.sha256()
        with open(path, "rb") as fh:
            for block in iter(lambda: fh.read(chunk), b""):
                h.update(block)
        return h.hexdigest()

    @staticmethod
    def _environment() -> dict:
        def sh(cmd):
            try:
                return subprocess.run(cmd, capture_output=True, text=True, check=True).stdout.strip()
            except Exception:
                return None
        import geopandas, shapely, pyproj
        return {"git_rev": sh(["git", "rev-parse", "HEAD"]),
                "git_dirty": bool(sh(["git", "status", "--porcelain", "--untracked-files=no"])),
                "python": sys.version.split()[0], "hostname": platform.node(),
                "user": os.environ.get("USER") or os.environ.get("USERNAME"),
                "packages": {"geopandas": geopandas.__version__,
                             "shapely": shapely.__version__,
                             "pyproj": pyproj.__version__}}

    def parameters(self, **kwargs):
        self.record["parameters"].update({k: str(v) for k, v in kwargs.items()})

    def input(self, path, gdf=None, hash_it=True):
        path = Path(path)
        entry = {"path": str(path.resolve()), "bytes": path.stat().st_size,
                 "sha256": self._sha256(path) if hash_it else None}
        if gdf is not None:
            entry.update({"features": len(gdf), "crs": str(gdf.crs),
                          "bounds": [round(float(v), 3) for v in gdf.total_bounds]})
        self.record["inputs"].append(entry)
        return entry

    def output(self, path, gdf=None, hash_it=True):
        path = Path(path)
        entry = {"path": str(path.resolve()), "bytes": path.stat().st_size,
                 "sha256": self._sha256(path) if hash_it else None}
        if gdf is not None:
            entry.update({"features": len(gdf), "crs": str(gdf.crs),
                          "bounds": [round(float(v), 3) for v in gdf.total_bounds]})
        self.record["outputs"].append(entry)
        return entry

    @contextmanager
    def step(self, name, **params):
        started = time.perf_counter()
        entry = {"name": name, "params": {k: str(v) for k, v in params.items()}}
        self.record["steps"].append(entry)
        try:
            yield entry
            entry["status"] = "ok"
        except Exception as exc:
            entry["status"] = "failed"
            entry["error"] = f"{type(exc).__name__}: {exc}"
            self.record["status"] = "failed"
            raise
        finally:
            entry["duration_s"] = round(time.perf_counter() - started, 3)

    def finish(self, status="ok") -> Path:
        self.record["status"] = status if self.record["status"] != "failed" else "failed"
        self.record["duration_s"] = round(time.perf_counter() - self.started, 2)
        self.record["finished_utc"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
        self.out_dir.mkdir(parents=True, exist_ok=True)
        dest = self.out_dir / f"{self.run_id}.json"
        dest.write_text(json.dumps(self.record, indent=2), encoding="utf-8")
        with open(self.out_dir.parent / "runs.jsonl", "a", encoding="utf-8") as fh:
            fh.write(json.dumps({k: self.record[k] for k in
                                 ("run_id", "pipeline", "started_utc", "duration_s", "status")}) + "\n")
        return dest

import geopandas as gpd
from lineage import RunRecorder

rec = RunRecorder("parcels_clean")
rec.parameters(target_crs="EPSG:27700", drop_invalid=True, buffer_m=0)

with rec.step("read") as s:
    gdf = gpd.read_file("data/raw/parcels.gpkg")
    rec.input("data/raw/parcels.gpkg", gdf)
    s["rows_out"] = len(gdf)

with rec.step("clean") as s:
    s["rows_in"] = len(gdf)
    gdf = gdf[gdf.geometry.is_valid & gdf.geometry.notna()].to_crs("EPSG:27700")
    s["rows_out"] = len(gdf)

with rec.step("write") as s:
    gdf.to_file("data/out/parcels_clean.gpkg", driver="GPKG")
    rec.output("data/out/parcels_clean.gpkg", gdf)

print("record:", rec.finish())
print(json.dumps([{k: v for k, v in step.items()} for step in rec.record["steps"]]))

record: logs/runs/20261010T223853Z-4d914e.json
[{"name": "read", "params": {}, "rows_out": 21253, "status": "ok", "duration_s": 0.161}, {"name": "clean", "params": {}, "rows_in": 21253, "rows_out": 21253, "status": "ok", "duration_s": 0.06}, {"name": "write", "params": {}, "status": "ok", "duration_s": 0.568}]

Example 2: trace an output back to its sources

import json
from pathlib import Path

def trace(output_path: str, runs_dir=Path("logs/runs"), depth=0):
    """Walk backwards from an output to the inputs that produced it."""
    for record_file in sorted(runs_dir.glob("*.json"), reverse=True):
        rec = json.loads(record_file.read_text())
        if any(o["path"].endswith(output_path) for o in rec["outputs"]):
            pad = "  " * depth
            print(f"{pad}{output_path}")
            print(f"{pad}  ← run {rec['run_id']} ({rec['started_utc']}, git {(rec['environment']['git_rev'] or 'none')[:8]})")
            for inp in rec["inputs"]:
                print(f"{pad}  ← input {Path(inp['path']).name} "
                      f"({inp.get('features', '?')} features, {inp['sha256'][:12]}…)")
                trace(Path(inp["path"]).name, runs_dir, depth + 2)
            return
    print("  " * depth + f"{output_path} (no run record — external source)")

trace("parcels_clean.gpkg")
trace("parcels_published.gpkg")

parcels_clean.gpkg
  ← run 20261010T223853Z-4d914e (2026-10-10T22:38:53+00:00, git 6bfec3f2)
  ← input parcels.gpkg (21253 features, 1580163716b0…)
    parcels.gpkg (no run record — external source)
parcels_published.gpkg (no run record — external source)

Example 3: detect when an input changed

import json
from pathlib import Path

def input_changed_since_last_run(path: Path, pipeline: str, runs_dir=Path("logs/runs")) -> bool:
    current = RunRecorder._sha256(path)
    for record_file in sorted(runs_dir.glob("*.json"), reverse=True):
        rec = json.loads(record_file.read_text())
        if rec.get("pipeline") != pipeline:
            continue
        for inp in rec["inputs"]:
            if Path(inp["path"]).name == path.name:
                return inp["sha256"] != current
    return True          # never seen before

if not input_changed_since_last_run(Path("data/raw/parcels.gpkg"), "parcels_clean"):
    print("input unchanged — skipping run")           # in a script: raise SystemExit(0)
else:
    print("input changed — running")

input unchanged — skipping run

Lineage records double as a cache key, which turns an audit feature into a performance feature.

Example 4: publish lineage as a small report

import json
from pathlib import Path
import pandas as pd

records = [json.loads(p.read_text()) for p in sorted(Path("logs/runs").glob("*.json"))]
df = pd.DataFrame([{
    "run_id": r["run_id"],
    "started": r["started_utc"],
    "status": r.get("status", "?"),                  # older records may lack a field
    "duration_s": r.get("duration_s"),
    "in_features": sum(i.get("features", 0) for i in r["inputs"]),
    "out_features": sum(o.get("features", 0) for o in r["outputs"]),
    "git": (r["environment"].get("git_rev") or "")[:8],
    "dirty": r["environment"].get("git_dirty"),
} for r in records])

df["retained_pct"] = (df["out_features"] / df["in_features"] * 100).round(1)
print(df.tail(12).to_string(index=False))
df.to_csv("logs/run_history.csv", index=False)

                 run_id                   started status  duration_s  in_features  out_features      git dirty  retained_pct
       20261010T223851Z 2026-10-10T22:38:51+00:00     ok        0.51        21253         21253 6bfec3f2  None         100.0
20261010T223853Z-4d914e 2026-10-10T22:38:53+00:00     ok        0.79        21253         21253 6bfec3f2 False         100.0

A retention percentage per run is a remarkably good health metric: a step that quietly starts dropping half its rows shows up as a column that moves.

Three questions that arrive weeks later

Lineage answers three questions that arrive weeks after the work: what produced this, from what, and would it produce the same thing today. None of them can be answered from a file on disk, because a file records its content and nothing about its history.

The unit that makes it work is the run: one execution, one id, one record. Everything else hangs off that id — log lines, output filenames, a column in the data, an entry in the ledger. When someone asks about a layer produced in March, the id in its metadata takes you straight to the record, and the record's input checksums take you to the previous run that produced those, and so on back to the original download.

Checksums are what make the chain trustworthy. Paths are reused, filenames lie, and modification times change when a file is copied. A SHA-256 identifies the exact bytes, so "the input was the same file we used last month" becomes a fact rather than an assumption — and the same property makes the record usable as a cache key, letting a pipeline skip work whose inputs have not changed.

Environment capture deserves the same weight as data capture. Geospatial results depend on the versions of GDAL, GEOS and PROJ as much as on your code: PROJ's coordinate-operation database changes between releases, and a transformation can shift by metres. Recording those versions alongside the git revision — and whether the working tree was dirty — is the difference between a result you can reproduce and one you can only re-run and hope.

Per-step counts are the operational payoff. A pipeline that records rows in and rows out at every stage makes shrinkage visible immediately, and turns "the output looks wrong" into "the dedupe step dropped 40% instead of 4%, starting on the 14th". That is a question you can answer in a minute instead of a day, which is why this small amount of bookkeeping repays itself so quickly.

Slow hashes and secrets in metadata

• Hashing large inputs is slow: Use size-plus-mtime as a fingerprint for multi-gigabyte working files and full SHA-256 for deliverables.
• Do not record secrets: Log the database name and host, never the connection string with the password.
• git rev-parse fails outside a repo: Guard it and fall back to a version string or an image tag baked in at build time.
• Wall-clock is not monotonic: Use time.perf_counter() for durations and datetime.now(timezone.utc) for timestamps; do not derive one from the other.
• JSON Lines beats one big JSON file: Appending is atomic-ish and the file stays readable if a run is killed mid-write.
• Stamp the data as well as the folder: A run_id column survives copying, format conversion and re-zipping, which a sidecar file does not.
• Retention: Run records are tiny; keep years of them. Rotating them away removes exactly the history you will want.

Internal links

• How to Make a GIS Workflow Reproducible in Python
• How to Build a Repeatable Data-Cleaning Report in GeoPandas
• How to Cache Pipeline Steps So Only Changed Data Is Reprocessed
• How to Log and Summarise Errors in a Batch GIS Job in Python
• How to Package GIS Deliverables into Zipped Bundles with Python
• My GIS Pipeline Fails Silently: Fixing Swallowed Errors and Wrong Exit Codes
• How to record lineage automatically in a pipeline — a run recorder that cannot be forgotten

FAQ

What is the minimum lineage worth recording?

Run id, timestamp, git revision, input paths with checksums, resolved parameters, and output paths with feature counts. That fits in forty lines of code and answers most questions you will be asked.

Where should the record live?

Write one JSON file per run under logs/runs/, append a one-line summary to a JSON Lines ledger, and stamp the run id into the output data itself so the two cannot be separated.

Should I hash every input?

Hash deliverables and reference data. For large working files, a size-plus-mtime fingerprint is usually enough and far faster; upgrade to a full hash when the record has to be evidence.

How do I record which library versions produced a result?

Capture geopandas, shapely, pyproj and rasterio versions plus rasterio.__gdal_version__ and pyproj.proj_version_str. PROJ changes can move coordinates, so its version matters as much as your code's.

Can lineage records make the pipeline faster?

Yes. If the recorded input checksums match the current ones, nothing upstream has changed and the step can be skipped — the same record that serves an audit serves as a cache key.

How do I trace an output back to its original source?

Look up the run record that lists the output, read its inputs, then find the runs that produced those, recursively. Storing input checksums makes each hop unambiguous.

Do I need a lineage tool like OpenLineage or DataHub?

Not for a single pipeline. JSON records plus a JSONL ledger cover it. Reach for a platform when many pipelines share datasets and people outside your team need to query the graph.