Tracking Spatial Data Lineage with OpenLineage

You have a Python spatial validation job that reads a dataset, runs quality rules, and writes a validated output, and you need a durable, queryable record of what happened — which dataset version went in, which rule set ran, and what quality it produced — that survives long after the process exits. This page shows exactly how to emit OpenLineage events from that job: naming the input and output datasets, attaching custom facets for coordinate reference system (CRS), feature count, and validity rate, wiring the emitter into an Airflow or plain-Python task, and viewing the resulting graph in Marquez. It is the hands-on companion to Observability and Lineage for Spatial Validation, which frames where lineage fits among logs, metrics, and traces.

Prerequisites

• Python 3.10+ and openlineage-python==1.19.* (pip install "openlineage-python==1.19.*"). The 1.x line uses the event_v2 module and the OpenLineageClient shown below; older 0.x snippets on the web use a different import path that will not run.
• A running OpenLineage receiver. The simplest is Marquez via its Docker quickstart, which exposes an HTTP API on port 5000 and a UI on 3000. Set the environment variable OPENLINEAGE_URL=http://localhost:5000 so OpenLineageClient.from_environment() finds it. For offline testing, set OPENLINEAGE_URL to a console transport instead and events print to stdout.
• GeoPandas 0.14+ to compute the facet values (feature count, CRS, validity rate) from your validation run. This page assumes you already produce a per-feature validity result — for example from a GeoPandas rule engine.
• Gotcha — namespaces are load-bearing. OpenLineage identifies a dataset by the pair (namespace, name), not by name alone. Pick a namespace convention up front (for example the storage scheme: gpkg, postgres://host, s3://bucket) and use it consistently, or the same physical dataset appears as two disconnected nodes in the graph.

Step-by-Step Procedure

Step 1 — Point the client at Marquez

import os
from openlineage.client import OpenLineageClient

# OpenLineageClient.from_environment() reads OPENLINEAGE_URL (and optional
# OPENLINEAGE_API_KEY) — set them before the process starts.
os.environ.setdefault("OPENLINEAGE_URL", "http://localhost:5000")
client = OpenLineageClient.from_environment()

PRODUCER = "https://spatial-data-validation.com/validation-pipeline"
JOB_NAMESPACE = "spatial-validation"

Verification: run curl -s http://localhost:5000/api/v1/namespaces and confirm it returns JSON. If the connection is refused, Marquez is not up and every client.emit() call will raise a connection error.

Step 2 — Emit a START event

from datetime import datetime, timezone
from openlineage.client.event_v2 import RunEvent, RunState, Run, Job, Dataset
from openlineage.client.uuid import generate_new_uuid

def utcnow() -> str:
    return datetime.now(timezone.utc).isoformat()

run_id = str(generate_new_uuid())
input_ds = Dataset(namespace="gpkg", name="parcels")

client.emit(RunEvent(
    eventType=RunState.START,
    eventTime=utcnow(),
    run=Run(runId=run_id),
    job=Job(namespace=JOB_NAMESPACE, name="parcel_validation"),
    inputs=[input_ds],
    outputs=[],
    producer=PRODUCER,
))

Verification: open the Marquez UI, select the spatial-validation namespace, and confirm a parcel_validation job now shows one run in the RUNNING state with parcels as an input.

Step 3 — Define custom spatial facets

A facet is a plain dictionary with two required keys — _producer (a URI identifying who emitted it) and _schemaURL (a URI for the facet’s schema) — plus your own fields. Compute the spatial values from the validation result.

import geopandas as gpd

def build_spatial_facet(gdf: gpd.GeoDataFrame, valid_mask) -> dict:
    """Compose a custom run facet describing the spatial quality outcome."""
    feature_count = len(gdf)
    validity_rate = float(valid_mask.mean()) if feature_count else 0.0
    minx, miny, maxx, maxy = gdf.total_bounds
    return {
        "spatialQuality": {
            "_producer": PRODUCER,
            "_schemaURL": PRODUCER + "/facets/spatial-quality.json",
            "crs": str(gdf.crs),                       # e.g. "EPSG:4326"
            "featureCount": feature_count,
            "validityRate": round(validity_rate, 4),   # e.g. 0.9812
            "geometryTypes": sorted(gdf.geom_type.dropna().unique().tolist()),
            "boundingBox": [minx, miny, maxx, maxy],
        }
    }

Verification: call build_spatial_facet(gdf, valid_mask) on a small sample and print it. Confirm _producer and _schemaURL are present — Marquez rejects facets missing either key, and the event is silently dropped.

Step 4 — Emit a COMPLETE event with the facet attached

output_ds = Dataset(namespace="gpkg", name="parcels_validated")
facets = build_spatial_facet(gdf, valid_mask)

client.emit(RunEvent(
    eventType=RunState.COMPLETE,
    eventTime=utcnow(),
    run=Run(runId=run_id, facets=facets),
    job=Job(namespace=JOB_NAMESPACE, name="parcel_validation"),
    inputs=[input_ds],
    outputs=[output_ds],
    producer=PRODUCER,
))

Verification: refresh the Marquez run view. The run should now read COMPLETED, the graph should show an edge from parcels through parcel_validation to parcels_validated, and the run detail panel should list the spatialQuality facet with your validityRate and crs values.

Step 5 — Handle failures with a FAIL event

Never let an exception leave a run stranded in START. Wrap the whole run so a failure emits a terminal event.

def run_with_lineage(gdf, validate_fn):
    run_id = str(generate_new_uuid())
    input_ds = Dataset(namespace="gpkg", name="parcels")
    job = Job(namespace=JOB_NAMESPACE, name="parcel_validation")

    client.emit(RunEvent(eventType=RunState.START, eventTime=utcnow(),
                         run=Run(runId=run_id), job=job,
                         inputs=[input_ds], outputs=[], producer=PRODUCER))
    try:
        valid_mask = validate_fn(gdf)
        facets = build_spatial_facet(gdf, valid_mask)
        client.emit(RunEvent(
            eventType=RunState.COMPLETE, eventTime=utcnow(),
            run=Run(runId=run_id, facets=facets), job=job,
            inputs=[input_ds],
            outputs=[Dataset(namespace="gpkg", name="parcels_validated")],
            producer=PRODUCER))
        return valid_mask
    except Exception as exc:
        client.emit(RunEvent(
            eventType=RunState.FAIL, eventTime=utcnow(),
            run=Run(runId=run_id, facets={"errorMessage": {
                "_producer": PRODUCER,
                "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/ErrorMessageRunFacet.json",
                "message": str(exc)}}),
            job=job, inputs=[input_ds], outputs=[], producer=PRODUCER))
        raise

Verification: pass a validate_fn that raises, and confirm the Marquez run ends in the FAILED state carrying the error message facet — not stuck on RUNNING.

Step 6 — Wire the emitter into an Airflow task

Any orchestrator works; the pattern is to call the helper from the task body. In an Airflow directed acyclic graph (DAG), a PythonOperator invokes run_with_lineage so lineage publishes on every scheduled run.

from airflow import DAG
from airflow.operators.python import PythonOperator
import geopandas as gpd
import pendulum

def validate_task(**context):
    gdf = gpd.read_file("/data/parcels.gpkg")
    run_with_lineage(gdf, validate_fn=lambda g: g.geometry.is_valid)

with DAG(
    dag_id="parcel_validation",
    start_date=pendulum.datetime(2026, 7, 1, tz="UTC"),
    schedule="@daily",
    catchup=False,
) as dag:
    PythonOperator(task_id="validate_parcels", python_callable=validate_task)

Verification: trigger the DAG once and confirm a new parcel_validation run appears in Marquez per execution date. (Airflow also ships a native OpenLineage provider that emits task-level lineage automatically; the explicit emitter here gives you control over the custom spatial facets it does not know about.)

Interpreting Results

In the Marquez graph, each dataset is a node and each job an edge-producing node between its inputs and outputs. A healthy validation job produces a stable two-node-plus-job pattern: parcels → parcel_validation → parcels_validated. Click any run to inspect its facets — the validityRate on the spatialQuality facet is your at-a-glance quality reading for that specific dataset version, and crs confirms the run operated in the projection you expected.

The run state itself is diagnostic. A steady stream of COMPLETE runs with a slowly declining validityRate signals gradual upstream data-quality erosion. A FAIL run with an errorMessage facet tells you the job broke and why, without opening application logs. And the timeline of runs against one dataset gives stewards the reproducibility record compliance frameworks require: for any past date, the graph shows which rule set version ran and what validity it observed.

Gotchas & Edge Cases

Facets missing _producer or _schemaURL are silently dropped. OpenLineage validates every facet against those two required keys. If you omit them, the event may still post but the facet vanishes, and you will spend an afternoon wondering why your validityRate never appears in Marquez. Always include both.

Inconsistent dataset namespaces fork the graph. Emitting parcels under namespace gpkg on one run and file on another creates two unrelated nodes for the same physical dataset, breaking the lineage chain. Centralise namespace construction in one helper so it is impossible to drift.

eventTime must be timezone-aware ISO 8601. A naive datetime.now().isoformat() (no timezone) is accepted by some backends and rejected by others, and it corrupts run ordering across hosts in different zones. Always use datetime.now(timezone.utc).isoformat().

Emitting synchronously on the critical path can stall the job. client.emit() makes a blocking HTTP call by default. If Marquez is slow or briefly unreachable, your validation task blocks with it. For production, emit from a background thread or accept that lineage is best-effort and wrap emit() so a transport error is logged but never fails the validation run.

A run_id must be unique per execution, not per job. Reusing one run_id across scheduled runs collapses them into a single run in Marquez and destroys the timeline. Mint a fresh UUID at the start of every execution.

When to Escalate

The single-job emission pattern here covers one validation task producing one output. Move to a heavier approach when:

• You need column-level and multi-hop lineage across a whole pipeline. When validation is one of many transforms and you need to trace a field from raw ingest through several jobs, adopt an orchestrator-native OpenLineage integration (the Airflow, Dagster, or Spark providers) that emits lineage for every task automatically, and reserve custom facets for the spatial specifics. The broader design trade-offs live in Observability and Lineage for Spatial Validation.
• Lineage must satisfy a formal audit or regulatory obligation. If the provenance record is evidence for an INSPIRE (Infrastructure for Spatial Information in Europe) or ISO (International Organization for Standardization) 19157 conformance claim, the ad-hoc facets here are a starting point but not the whole story — you need retention, access control, and a documented mapping from facets to metadata elements, covered in Compliance Framework Alignment.
• Emission volume threatens the validation throughput. High-frequency streaming validation emitting a lineage event per micro-batch can overwhelm a single Marquez instance; move to buffered, asynchronous transport and coarser job granularity as described in the parent Validation Pipeline Architecture.

Frequently Asked Questions

Related:

• Observability and Lineage for Spatial Validation — where lineage fits alongside logs, metrics, and traces
• Compliance Framework Alignment — turning lineage records into audit evidence for INSPIRE and ISO 19157
• Building Rule Engines with GeoPandas — the validation logic that produces the validity results these facets report

Back to Observability and Lineage for Spatial Validation