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:
- capture inputs — path, size, mtime, checksum — before processing
- capture the environment — git revision, Python and key library versions
- capture parameters — the resolved config, not the defaults you assume
- capture outputs — counts, bounds, CRS, checksum
- 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")
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),
"code": {"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)},
}
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"]))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
Step-by-step solution
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() # e.g. 20260811T023014Z-9f2ab1A 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"]),
}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)}"Record the environment that produced the result
import platform, subprocess, 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"])),
"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"),
}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. 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_")},
}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)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://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 sideAdding columns works too, and survives any format conversion:
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
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())JSON Lines appends cheaply and reads as a DataFrame, which is all the infrastructure most pipelines ever need.
Code examples
Example 1: a complete lineage recorder
"""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"])),
"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 destimport 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())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'][: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_published.gpkg")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")
raise SystemExit(0)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["status"],
"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"]["git_rev"] or "")[:8],
"dirty": r["environment"]["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)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.
Explanation
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.
Edge cases or notes
- 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-parsefails 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 anddatetime.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_idcolumn 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
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.