Tracking Data Lineage Across Geospatial ETL Steps

Threading lineage through a geospatial extract-transform-load pipeline means recording, at every hop, exactly what went in, what operation ran, and what came out. When each ETL step captures an input hash, its operation metadata, and the identifiers of the features it produces, the resulting chain lets a compliance officer trace any output polygon back to the raw survey file that spawned it. This guide builds that chain concretely, storing it as a JSON lineage record that a graph store or audit reviewer can walk end to end.

Prerequisites

Step-by-step

Step 1: Fingerprint each extracted input

Lineage begins at extraction. Hash the raw bytes of every source before any parsing or repair alters them, and seed a record keyed by a unique run id. The digest is what makes the chain reconstructable later.

import hashlib, uuid, json
from datetime import datetime, timezone
import geopandas as gpd

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

def start_lineage(parcel_path: str) -> dict:
    gdf = gpd.read_file(parcel_path)
    return {
        "run_id": str(uuid.uuid4()),
        "created": datetime.now(timezone.utc).isoformat(),
        "nodes": [{
            "node_id": "input:parcels",
            "kind": "source",
            "sha256": sha256_of(parcel_path),
            "crs": str(gdf.crs),   # record the CRS exactly as delivered
            "features": int(len(gdf)),
        }],
        "edges": [],
    }

Step 2: Record operation metadata per transform

Each transform appends a node describing what it did and an edge linking it to the node it consumed. Capturing the CRS inside the operation metadata is essential, because a distance measured under the wrong projection is silently wrong. Always project to a linear CRS before any measurement.

def add_step(lineage: dict, node_id: str, op: str, params: dict,
             crs: str, from_node: str) -> str:
    lineage["nodes"].append({
        "node_id": node_id,
        "kind": "transform",
        "operation": op,
        "params": params,
        "crs": crs,            # projection in force for this operation
    })
    lineage["edges"].append({"from": from_node, "to": node_id})
    return node_id

def reproject(lineage, gdf, metric_crs="EPSG:32617"):
    projected = gdf.to_crs(metric_crs)  # UTM 17N, metres — measure only after this
    add_step(lineage, "op:reproject", "to_crs",
             {"target": metric_crs}, metric_crs, "input:parcels")
    return projected

Step 3: Capture output ids after a spatial operation

When a step generates geometry, such as a compliance buffer, record the identifiers of the features it emitted so downstream consumers can be matched back to their producer. The buffer runs in the metric CRS established in the previous step.

def buffer_step(lineage, gdf, distance_m, crs="EPSG:32617"):
    out = gdf.copy()
    out["geometry"] = out.geometry.buffer(distance_m)  # metric CRS already set
    node = add_step(lineage, "op:buffer", "buffer",
                    {"distance_m": distance_m}, crs, "op:reproject")
    # attach the concrete output feature ids produced by this node
    lineage["nodes"][-1]["output_ids"] = out["parcel_id"].astype(str).tolist()
    return out, node

Step 4: Serialize the lineage record

Close the record with a deterministic JSON dump. Sorted keys make two runs on identical data diff cleanly, which is invaluable when proving that nothing changed between evaluations.

def write_lineage(lineage: dict, path: str) -> None:
    lineage["finished"] = datetime.now(timezone.utc).isoformat()
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(lineage, fh, indent=2, sort_keys=True)

Verification

Confirm the chain is complete before trusting it. Every edge must reference nodes that actually exist, and every transform that measured distance must carry a metric CRS. A quick structural check catches orphaned references and geographic-CRS mistakes early.

def verify_lineage(lineage: dict) -> None:
    ids = {n["node_id"] for n in lineage["nodes"]}
    for e in lineage["edges"]:
        assert e["from"] in ids and e["to"] in ids, f"dangling edge {e}"
    for n in lineage["nodes"]:
        if n.get("operation") in {"buffer", "sjoin_nearest"}:
            assert "326" in n["crs"] or "269" in n["crs"], "measured under non-metric CRS"
    print(f"lineage OK: {len(lineage['nodes'])} nodes, {len(lineage['edges'])} edges")

Spot-check one output feature by walking its edges backward to the source node; the SHA-256 you land on should match the original file on disk.

Common Pitfalls

  • Hashing the GeoDataFrame instead of the file. In-memory row order and dtype coercion vary between runs, so a hash of the parsed frame is unstable. Digest the raw source bytes.
  • Losing the ingest CRS after reprojection. Once you call to_crs, the original authority is gone from the active frame. Capture it in the source node during extraction, not later.
  • Recording steps but never the output ids. Without emitted identifiers, you can prove an operation ran but cannot tie a specific output polygon back to it, defeating the audit purpose.

Frequently Asked Questions

How is data lineage different from a simple pipeline log?

A conventional log narrates events in time order and is usually discarded after debugging. A lineage record is a structured graph of data dependencies designed to be kept: it links specific inputs to specific outputs through named operations, so any result can be reconstructed. The parent module on provenance and lineage tracking treats this graph as a permanent system of record rather than a transient diagnostic.

Should lineage records live in JSON files or a graph database?

Both are valid. JSON on append-only storage is simple, portable, and diff-friendly, which suits per-run audit files. A graph database such as Neo4j shines when you need to query across thousands of runs, for example to find every result that consumed a since-corrected parcel. Many teams write JSON per run and periodically load those records into a graph for cross-run analysis.

Does capturing lineage slow the pipeline noticeably?

The dominant cost is hashing source files, which is linear in file size and typically negligible next to reprojection and overlay operations. Appending node and edge dictionaries is effectively free. For very large inputs, hash while streaming during extraction so the read is not repeated.

How do I record which rule produced a compliance verdict?

Attach an immutable rule reference to the lineage record at close time, resolving it to a commit SHA or semantic version rather than a branch name. The companion guide on versioning rule references in audit trails covers exactly how to pin and store that reference.

Summary

By fingerprinting inputs at extraction, annotating every transform with its parameters and CRS, and recording the output ids each step emits, a geospatial ETL pipeline produces a lineage record that reconstructs any result on demand. Serialized deterministically to JSON and verified for structural completeness, that record becomes the evidentiary backbone the wider provenance and lineage tracking module relies on to keep automated compliance defensible.