How to record lineage automatically in a pipeline
On this page
Problem statement
Lineage that a person writes is lineage that stops being written the week the deadline moves. The only records that stay accurate are the ones the pipeline produces as a side effect of doing the work, from the same variables that did it.
That means three things have to happen in the code rather than in a document: inputs get registered when they are opened, steps get registered when they run, and the record gets written in the same function that writes the output. Everything else — the environment, the code version, the timestamps — is free.
This guide builds that as a small context manager, wires it into a pipeline, and makes the resulting store queryable.
Quick answer
import json, pathlib, datetime, subprocess, sys, hashlib
import importlib.metadata as md
class Run:
def __init__(self, output, store="lineage"):
self.output, self.store = str(output), pathlib.Path(store)
self.rec = {"output": str(output), "inputs": [], "steps": [],
"started": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")}
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
self.rec["status"] = "failed" if exc else "ok"
if exc:
self.rec["error"] = f"{exc_type.__name__}: {exc}"
self.rec["finished"] = datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")
self.rec["environment"] = environment()
if not exc and pathlib.Path(self.output).exists():
self.rec["output_sha256"] = file_hash(self.output)
self.store.mkdir(parents=True, exist_ok=True)
name = pathlib.Path(self.output).name + f".{self.rec['started'].replace(':', '')}.json"
(self.store / name).write_text(json.dumps(self.rec, indent=2))
return False # never swallow the exception
def input(self, name, path, **kw):
self.rec["inputs"].append({"name": name, "path": str(path),
"sha256": file_hash(path), **kw})
def step(self, op, **params):
self.rec["steps"].append({"op": op, "params": params,
"at": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")})The two design decisions that make it work: it records failures as well as successes, and it never swallows the exception.
What you get — from The data used on this page
(933, 933) DEM cells | True
One record per output, register inputs, hash as on disk
1. Scope the record to one output
One record per artefact, named after it. That is the unit people ask about.
2. Register inputs where they are opened
Wrap the reader, so an input cannot enter the pipeline unrecorded:
def read_vector(run, name, path, **kw):
run.input(name, path, **kw)
return gpd.read_file(path)
def read_raster(run, name, path, **kw):
run.input(name, path, **kw)
with rasterio.open(path) as src:
return src.read(1, masked=True), src.profile3. Hash inputs as they are on disk
For files you received, the byte hash is the right thing and is cheap. For large inputs, hash once and cache by path plus mtime plus size.
4. Record parameters that change the answer
The threshold, the resampling method, the CRS, the tolerance, the seed. Not the chunk size or the worker count.
5. Capture the environment once, at the end
Python version, the library versions that do the arithmetic, the git commit and whether the tree was dirty.
6. Record failures
A run that raised still explains a gap in a series, and a record with status: failed and the error is the fastest way to answer "why is there no output for the 14th?".
7. Write to a store, not only beside the output
A directory of records lets you ask questions across runs — which outputs used a source that has turned out to be wrong, which runs used the library version that had the bug.
8. Put the record in the delivery
A copy beside the artefact travels with it; the store stays behind.
Helpers and a pipeline that records itself
The data used on this page
The Copernicus GLO-30 DEM over Rutland (© DLR e.V. 2010–2014 and © Airbus Defence and Space GmbH 2014–2018, provided under COPERNICUS by the European Union and ESA), reprojected to British National Grid, and the county's boundary from the ONS December 2024 district boundaries (OGL). The working directory is made a git repository with one commit, so the records have a commit to name.
import json
import subprocess
from pathlib import Path
import geopandas as gpd
import rasterio
import requests
from osgeo import gdal
gdal.UseExceptions()
for folder in ("data", "out"):
Path(folder).mkdir(exist_ok=True)
if not Path("data/dem_27700.tif").exists():
gdal.Warp("data/dem_27700.tif", "/vsicurl/https://copernicus-dem-30m.s3.amazonaws.com/"
"Copernicus_DSM_COG_10_N52_00_W001_00_DEM/Copernicus_DSM_COG_10_N52_00_W001_00_DEM.tif",
dstSRS="EPSG:27700", outputBounds=(479000, 292000, 507000, 320000), xRes=30, yRes=30,
resampleAlg="bilinear", dstNodata=-9999)
Path("data/rutland.geojson").write_bytes(requests.get(
"https://services1.arcgis.com/ESMARspQHYMw9BZ9/arcgis/rest/services/"
"Local_Authority_Districts_December_2024_Boundaries_UK_BFC/FeatureServer/0/query",
params={"where": "LAD24NM='Rutland'", "outFields": "LAD24NM", "outSR": 27700, "f": "geojson"}, timeout=120).content)
if not Path(".git").exists():
Path(".gitignore").write_text("*\n!.gitignore\n")
for command in (["git", "init", "-q"], ["git", "add", ".gitignore"],
["git", "-c", "user.name=pipeline", "-c", "user.email=pipeline@example.org", "commit", "-qm", "pipeline"]):
subprocess.run(command, check=True)
for old in Path("lineage").glob("*.json") if Path("lineage").exists() else []:
old.unlink()
print(rasterio.open("data/dem_27700.tif").shape, "DEM cells |", subprocess.run(
["git", "rev-parse", "--short", "HEAD"], capture_output=True, text=True).stdout.strip() != "")(933, 933) DEM cells | True
Example 1 — the helpers
import hashlib, subprocess, sys, pathlib
import importlib.metadata as md
def file_hash(path, chunk=1 << 20):
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 environment(packages=("geopandas", "shapely", "pyproj", "rasterio", "pandas", "numpy")):
def version(p):
try:
return md.version(p)
except md.PackageNotFoundError:
return None
try:
commit = subprocess.check_output(["git", "rev-parse", "--short", "HEAD"],
text=True, stderr=subprocess.DEVNULL).strip()
dirty = bool(subprocess.check_output(["git", "status", "--porcelain"],
text=True, stderr=subprocess.DEVNULL).strip())
except Exception:
commit, dirty = None, None
return {"python": sys.version.split()[0],
"packages": {p: version(p) for p in packages},
"git": {"commit": commit, "dirty": dirty}}
env = environment()
print(env["python"], {k: v for k, v in env["packages"].items() if v}, "| dirty:", env["git"]["dirty"])3.14.4 {'geopandas': '1.1.4', 'shapely': '2.1.2', 'pyproj': '3.8.0', 'rasterio': '1.5.1', 'pandas': '3.0.6', 'numpy': '2.5.3'} | dirty: False
Example 2 — a pipeline that records itself
A real, small pipeline: Rutland's high ground — land above 150 m on the Copernicus DEM — as polygons, clipped to the county. The pipeline code lives in its own git repository, committed, so the record can name the commit.
from rasterio.features import shapes
from shapely.geometry import shape
def high_ground(output, threshold_m=150, min_area_m2=10_000):
with Run(output) as run:
dem, profile = read_raster(run, "copernicus_glo30", "data/dem_27700.tif",
version="2021_1", licence="Copernicus DEM licence")
county = read_vector(run, "ons_lad_2024_rutland", "data/rutland.geojson", licence="OGL-UK-3.0")
run.step("threshold", threshold_m=threshold_m)
mask = (dem.filled(-9999) > threshold_m).astype("uint8")
run.step("polygonise", connectivity=4)
polygons = gpd.GeoDataFrame(geometry=[shape(g) for g, v in shapes(mask, mask=mask == 1,
transform=profile["transform"])],
crs=profile["crs"])
run.step("clip_and_filter", min_area_m2=min_area_m2)
polygons = gpd.clip(polygons, county.to_crs(polygons.crs))
polygons = polygons[polygons.area >= min_area_m2]
run.step("write", driver="GPKG", layer="high_ground")
polygons.to_file(output, layer="high_ground", driver="GPKG")
return len(polygons), polygons.area.sum() / 1e6
n, km2 = high_ground("out/high_ground_150m.gpkg")
print(f"ok run: {n} polygons of high ground, {km2:.1f} km²")
try:
high_ground("out/high_ground_bad.gpkg", min_area_m2="10000") # a string where a number belongs
except TypeError as exc:
print("failed run raised:", str(exc)[:70])
for record in sorted(Path("lineage").glob("*.json")):
rec = json.loads(record.read_text())
print(f"{record.name[:40]:42} {rec['status']:7} steps={[s['op'] for s in rec['steps']]} "
f"{rec.get('error', '')[:40]}")ok run: 9 polygons of high ground, 20.2 km²
failed run raised: Invalid comparison between dtype=float64 and str
high_ground_150m.gpkg.2026-09-30T175804+ ok steps=['threshold', 'polygonise', 'clip_and_filter', 'write']
high_ground_bad.gpkg.2026-09-30T175804+0 failed steps=['threshold', 'polygonise', 'clip_and_filter'] TypeError: Invalid comparison between dt
The failed run still left a record — with the three steps that ran, both inputs, and the first line of the error — and the exception still reached the caller. That is the context manager doing both of its jobs.
Example 3 — query the store
import json, pathlib, collections
def load_store(store="lineage"):
return [json.loads(p.read_text()) for p in pathlib.Path(store).glob("*.json")]
def outputs_using(name_or_hash, store="lineage"):
hits = []
for rec in load_store(store):
for inp in rec["inputs"]:
if name_or_hash in (inp["name"], inp.get("sha256")):
hits.append((rec["output"], rec["started"], rec.get("status")))
return sorted(hits, key=lambda h: h[1], reverse=True)
def library_versions_used(store="lineage"):
seen = collections.defaultdict(set)
for rec in load_store(store):
for pkg, ver in (rec.get("environment", {}).get("packages") or {}).items():
seen[pkg].add(ver)
return {k: sorted(v) for k, v in seen.items()}
print(outputs_using("copernicus_glo30"))
print({k: v for k, v in library_versions_used().items() if k in ("geopandas", "rasterio", "shapely")})
latest = max(load_store(), key=lambda r: r["started"])
print("git:", latest["environment"]["git"], "| inputs hashed:", [i["sha256"][:12] for i in latest["inputs"]])[('out/high_ground_150m.gpkg', '2026-09-30T17:58:04+00:00', 'ok'), ('out/high_ground_bad.gpkg', '2026-09-30T17:58:04+00:00', 'failed')]
{'geopandas': ['1.1.4'], 'shapely': ['2.1.2'], 'rasterio': ['1.5.1']}
git: {'commit': '18370cb', 'dirty': False} | inputs hashed: ['7ffce2ede31f', 'c5440398b063']
library_versions_used is the query nobody expects to need and everybody eventually does, when a result changes and nothing in the code did.
A context manager, and recording failures
Why a context manager rather than a decorator
The record has to be written whether the body succeeded or failed, and it has to see the output file after it was written. __exit__ runs in both cases and runs last, which is exactly the shape of the problem. A decorator on the pipeline function would work too; a decorator per step would not, because the steps do not know about the output.
Why recording failures matters more than it sounds
Most questions about a pipeline are about something that is missing, not something that is wrong. A store that contains only successes cannot answer them, and the person asking has to reconstruct what happened from logs that have rotated.
Why the input hash is the field that pays off
Paths lie: the file at /data/current/dtm.tif is not the file that was there in January. A content hash turns "we used the 2024 DTM" into a claim you can test against the file you still have, and it is what distinguishes a data change from a code change when a result moves.
Why the store needs to stay small enough to read
A JSON file per run is fine for thousands of runs and unmanageable for millions. When it grows, load the records into DuckDB or SQLite and query them there — the schema is stable and shallow, so it maps onto a table without much effort.
Large inputs, credentials and download URLs
- Hash large inputs once. Cache by path, size and mtime.
- Do not record credentials. Record the endpoint and the account, not the token.
- Record the URL and the fetch time for downloads. Remote data changes silently.
- A dirty git tree invalidates the commit. Record the flag and mean it.
- Manual steps are lineage too. Record "edited by hand in QGIS" rather than omitting it.
- Timestamps in UTC. Local times across a DST boundary are unsortable.
- Name records so they cannot collide. The quick answer names a record after the output and the start time to the second, so two runs of the same output within one second write the same file and the second overwrites the first. Add a run id — a UUID, or the process id — to the name when runs can be that close together.
- Rotate or index the store. A directory listing is not a query plan.
- Ship a copy with the artefact. The store does not travel.
Internal links
- Provenance and lineage explained for spatial workflows — what the record has to contain
- How to record run metadata and lineage in a GIS pipeline — the pipeline-side pattern
- How to checksum spatial datasets so you can prove they match — hashing outputs rather than inputs
- You cannot tell which data version produced a map — the failure this prevents
- Reproducible GIS environments explained — pinning what the record names
- How to validate pipeline inputs and outputs in Python — the checks that run alongside
- Observability for GIS pipelines — where lineage sits among logs and metrics
- Dataset versioning explained — the version the record refers to
FAQ
How do I record lineage without adding work to every script?
Use a context manager that captures the environment and writes the record on exit, plus wrapped readers that register inputs as they are opened. The only manual calls left are one step per decision.
Should I record failed runs?
Yes. Most questions are about a missing output, and a record with a status and an error answers them immediately.
What parameters belong in the record?
Ones that change the result: thresholds, resampling methods, tolerances, CRSs, seeds. Not chunk sizes or worker counts.
Where should records be stored?
Both beside the artefact, so they travel, and in a central store, so you can ask which outputs used a given input.
How do I find every output that used a bad source?
Query the store for records whose inputs include that name or content hash. That query is the main reason to keep records centrally.
Do I need to hash every input?
Every input you rely on. Hash large files once and cache the digest by path, size and modification time.