Tracking Data Lineage with OpenLineage for ESG Audits #

This guide is the implementation reference for emitting verifiable, spatial-aware lineage events out of an MRV pipeline — the operational layer that turns the principles in MRV data lineage and provenance tracking into machine-readable evidence inside the wider MRV Architecture & Carbon Accounting Fundamentals stack. Where the parent component frames what provenance must capture, this page walks the exact OpenLineage integration: how a raster clip, a vector re-projection, and an emission calculation each become a hashed, schema-enforced RunEvent that an auditor can replay byte-for-byte.

ESG verification bodies and voluntary carbon registries now mandate cryptographic-grade provenance for every emission factor, land-use polygon, and supplier activity record entering a carbon accounting pipeline. Traditional application logs and ad-hoc metadata tables collapse when spatial datasets undergo multi-stage transformations across distributed orchestration layers. The integration below depends on deterministic CRS alignment at ingestion so that the coordinates it records are honest, and it feeds the audit trail that carbon credit registry data integration submits — while threading through GHG Protocol Scope 3 spatial mapping, where supply-chain attribution is unverifiable unless each boundary intersection carries its own recorded origin.

Root Cause Analysis #

The problem OpenLineage solves is structural: in a distributed orchestration layer, the relationship between a transformation and its inputs lives only in the runtime memory of the task that ran it. Once that task exits, the binding evaporates unless it was explicitly serialized. Application logs record that something happened, but not the addressable inputs, the active projection, or a content hash of the output — so they cannot reconstruct a chain of custody. When a reported tCO₂e figure is challenged, the team is left re-deriving provenance from filenames and commit history, which no verifier accepts.

Three failure mechanisms recur when teams attempt traceability without a schema-enforced lineage protocol:

1. CRS decoupling across stages. A raster layer in EPSG:3857 clipped against a vector boundary in EPSG:4326 introduces area distortion that compounds across aggregation stages. Without a spatial facet recorded per stage, downstream consumers cannot validate that the projection was consistent, and an area-based emission calculation is silently invalidated while the manifest still reports a clean datum tag.
2. Temporal drift masking source substitution. When a primary observation window is quietly swapped for a fallback epoch, the output dataset looks identical but the tonnage now derives from a different acquisition period. Unless the temporal window is bound into the lineage event, the substitution is invisible.
3. Scope 3 aggregation flattening upstream origin. Supplier activity data merged with regional grid factors loses the per-source registry identifier when intermediate tables are overwritten. The final figure is numerically plausible but untraceable to the verified project boundary it claims.

Each failure is deterministic, and therefore preventable, by attaching a typed spatial contract to every dataset a task emits. The remaining sections install that contract.

Diagnostic Pipeline / Pre-Flight Validation #

Before any lineage event is emitted, the task must inspect its inputs and detect the failure conditions above — emitting a RunEvent for a mathematically invalid spatial state pollutes the audit record with confident-looking garbage. The pre-flight gate validates that every input declares an explicit, machine-readable CRS, that bounding boxes fall within geodetic bounds, and that the projection matches the pipeline’s canonical reference before transformation begins.

import json
import structlog

log = structlog.get_logger("mrv.lineage.preflight")


def validate_crs_alignment(input_crs: str, expected_crs: str, dataset: str) -> None:
    """Halt the task before emission if CRS alignment fails GHG boundary rules."""
    if not input_crs or ":" not in input_crs:
        log.error("crs_missing", dataset=dataset, declared=input_crs)
        raise ValueError(f"Dataset '{dataset}' has no machine-readable CRS tag")
    if input_crs != expected_crs:
        rejection = {
            "status": "BLOCKED",
            "dataset": dataset,
            "reason": f"CRS mismatch: {input_crs} != {expected_crs}",
            "compliance_rule": "GHG Protocol Spatial Boundary Alignment",
        }
        log.error("crs_mismatch", **rejection)
        raise ValueError(json.dumps(rejection))


def validate_bbox(bbox: dict, dataset: str) -> None:
    """Reject geometries outside WGS84 geodetic bounds before any spatial join."""
    if not all(k in bbox for k in ("minx", "miny", "maxx", "maxy")):
        raise ValueError(f"Dataset '{dataset}' bbox incomplete: {bbox}")
    if not (-180 <= bbox["minx"] < bbox["maxx"] <= 180):
        raise ValueError(f"Dataset '{dataset}' longitude out of bounds: {bbox}")
    if not (-90 <= bbox["miny"] < bbox["maxy"] <= 90):
        raise ValueError(f"Dataset '{dataset}' latitude out of bounds: {bbox}")
    log.info("bbox_validated", dataset=dataset, bbox=bbox)

Wiring this gate into Airflow’s on_failure_callback or a Prefect task guard ensures that lineage only records states that have already passed spatial validation. The CRS contract enforced here is the same one detailed in how to align WGS84 to a local CRS in Python, applied at the moment of emission rather than at ingestion.

Deterministic Transformation Logic #

OpenLineage’s core RunEvent schema captures generic inputs and outputs, but ESG verification requires a domain-specific extension. Standard facets lack fields for coordinate reference systems, bounding-box extents, temporal resolution, and registry identifiers, so the integration injects a custom spatialProvenance facet onto every output dataset. The facet is the verifiable spatial contract — it binds the transformation to a projection and an observation window, and seals that binding with a crsValidationHash that an auditor recomputes to detect tampering or silent drift.

{
  "spatialProvenance": {
    "_producer": "https://github.com/esg-mrv/lineage-facets",
    "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/CustomFacet.json",
    "crs": "EPSG:4326",
    "boundingBox": {"minx": -122.5, "miny": 37.0, "maxx": -121.8, "maxy": 37.9},
    "temporalWindow": {"start": "2023-01-01T00:00:00Z", "end": "2023-12-31T23:59:59Z"},
    "sourceRegistryId": "VCS-1842",
    "scope3Category": "Category 11",
    "calculationMethodology": "IPCC 2006 Tier 2",
    "crsValidationHash": "sha256:a1b2c3d4..."
  }
}

The emitter below builds that facet inside a START/COMPLETE event pair so that a failure between the two is itself recorded. It uses the openlineage-python client to dispatch events over HTTP to a Marquez (or compatible) backend, runs the pre-flight gate first, and logs every boundary with structlog for an audit-ready JSON trail.

import uuid
import datetime
import hashlib
import structlog
from openlineage.client import OpenLineageClient
from openlineage.client.run import (
    RunEvent, RunState, Run, Job, InputDataset, OutputDataset,
)
from openlineage.client.facet import DocumentationJobFacet

log = structlog.get_logger("mrv.lineage.emitter")


class SpatialLineageEmitter:
    """Emits CRS-aware OpenLineage events for one MRV transformation."""

    def __init__(self, namespace: str, expected_crs: str = "EPSG:4326"):
        self.namespace = namespace
        self.expected_crs = expected_crs
        # Reads OPENLINEAGE_URL / API key from the environment.
        self.client = OpenLineageClient.from_environment()

    def _crs_hash(self, crs_epsg: int, bbox: dict) -> str:
        """Immutable checksum binding projection to extent for replay validation."""
        payload = (
            f"EPSG:{crs_epsg}|{bbox.get('minx','')}|{bbox.get('miny','')}"
            f"|{bbox.get('maxx','')}|{bbox.get('maxy','')}"
        )
        return "sha256:" + hashlib.sha256(payload.encode()).hexdigest()

    def emit(self, task_id, inputs, outputs, crs_epsg, registry_id,
             scope3_cat, methodology, temporal_window):
        run_id = str(uuid.uuid4())
        now = lambda: datetime.datetime.now(datetime.timezone.utc).isoformat()

        # Validate every input BEFORE recording a START event.
        for i in inputs:
            validate_crs_alignment(i["crs"], self.expected_crs, i["name"])
            validate_bbox(i["bbox"], i["name"])

        bbox = inputs[0]["bbox"]
        spatial_facet = {
            "_producer": "https://github.com/esg-mrv/lineage-facets",
            "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/CustomFacet.json",
            "crs": f"EPSG:{crs_epsg}",
            "boundingBox": bbox,
            "temporalWindow": temporal_window,
            "sourceRegistryId": registry_id,
            "scope3Category": scope3_cat,
            "calculationMethodology": methodology,
            "crsValidationHash": self._crs_hash(crs_epsg, bbox),
        }

        job = Job(
            namespace=self.namespace, name=task_id,
            facets={"documentation": DocumentationJobFacet(
                description=f"ESG spatial transform: {task_id}")},
        )
        run = Run(runId=run_id, facets={})
        in_ds = [InputDataset(namespace=i["namespace"], name=i["name"]) for i in inputs]
        out_ds = [OutputDataset(
            namespace=o["namespace"], name=o["name"],
            facets={"spatialProvenance": spatial_facet}) for o in outputs]

        # START → COMPLETE pair: an exception between them leaves a FAIL event.
        self.client.emit(RunEvent(
            eventType=RunState.START, eventTime=now(), run=run, job=job,
            inputs=in_ds, producer=spatial_facet["_producer"]))
        try:
            self.client.emit(RunEvent(
                eventType=RunState.COMPLETE, eventTime=now(), run=run, job=job,
                inputs=in_ds, outputs=out_ds, producer=spatial_facet["_producer"]))
            log.info("lineage_emitted", task=task_id, run_id=run_id,
                     crs=spatial_facet["crs"], registry=registry_id)
        except Exception as exc:
            self.client.emit(RunEvent(
                eventType=RunState.FAIL, eventTime=now(), run=run, job=job,
                producer=spatial_facet["_producer"]))
            log.error("lineage_failed", task=task_id, run_id=run_id, error=str(exc))
            raise

The crsValidationHash acts as an immutable checksum for spatial alignment: when an auditor recomputes it from the recorded CRS and bounding box, any divergence proves the output was reprojected or substituted without re-emitting lineage. Downstream consumers reject any payload where the hash fails to match, preventing silent area-calculation errors from propagating into aggregation.

Compliance Gating & Audit Trail Generation #

Carbon credit registries (Verra, Gold Standard, ART) and GHG Protocol Scope 3 categories require explicit mapping between spatial boundaries and emission methodologies, and the spatialProvenance facet is what carries that mapping through every aggregation step. When supplier activity data merges with regional grid emission factors, the lineage event preserves the exact upstream registry ID, temporal coverage, and calculation tier so the final figure remains traceable to a verified project boundary — the same submission contract handled in integrating Verra and Gold Standard APIs into Python pipelines.

A compliance-ready aggregation facet carries four load-bearing fields:

• sourceRegistryId — links spatial polygons to verified carbon project boundaries (e.g. VCS-1842).
• scope3Category — maps to GHG Protocol categories such as Category 4 or Category 11, mirroring the attribution logic in the step-by-step Scope 3 geospatial calculation.
• calculationMethodology — specifies the IPCC tier, registry methodology ID, or custom emission-factor derivation.
• aggregationLogic — documents whether spatial weighting is area-proportional, population-weighted, or uniform.

During third-party verification, auditors query the lineage backend to reconstruct the exact execution graph for a reporting period. A compliant reconstruction workflow runs five steps:

1. Event retrieval. Fetch all RunEvent payloads matching the reporting namespace and temporal window from the backend (e.g. Marquez).
2. Facet extraction. Parse the spatialProvenance facet from each output dataset.
3. Chain validation. Recompute and compare crsValidationHash across sequential tasks, rejecting any event whose projection drift exceeds the tolerance defined in the GHG Protocol Corporate Standard.
4. Registry cross-reference. Match sourceRegistryId against official registry databases to confirm project validity and vintage alignment.
5. Methodology trace. Confirm calculationMethodology aligns with the disclosed framework (ISO 14064-3, SBTi FLAG, or CSRD ESRS E1).

This sequence eliminates manual spreadsheet reconciliation and yields a machine-readable trail from raw Sentinel-2 tiles to the final inventory total — the evidentiary completeness ISO 14064-3 §5.4 and CSRD ESRS E1 disclosures are scrutinized for.

Production Integration #

In production the emitter wraps each orchestration task so lineage is committed as a first-class output, not a side effect. The end-to-end execution pattern for a single MRV transformation follows six ordered stages:

1. Ingest — load source rasters and supplier geometries with their declared CRS and acquisition window attached as task parameters.
2. Diagnose — run validate_crs_alignment and validate_bbox over every input; route any BLOCKED record to a remediation queue before emission.
3. Transform — perform the raster clip, re-projection, or emission calculation on a single canonical pixel grid.
4. Validate — recompute crsValidationHash on the output and assert it against the expected projection.
5. Export — serialize the result as GeoParquet or a registry-ready GeoTIFF with the spatialProvenance facet embedded alongside.
6. Submit — emit the START/COMPLETE event pair to the collector and forward the registry payload downstream.

For continental-scale runs, emit one event per chunk rather than per scene: align lineage emission to the same tiling the pipeline already uses for out-of-core I/O, so each Dask partition produces an addressable, hashed artifact with its own facet. Batch the HTTP dispatch behind a buffered transport (the client’s async or kafka transport) to keep emission off the critical path, and back-pressure on collector failure rather than dropping events — a lost lineage record is an unverifiable artifact. Embedding this discipline at the orchestration layer is now a baseline control for enterprise MRV systems: it guarantees cryptographic traceability, prevents projection-induced calculation errors, and satisfies increasingly stringent ESG verification mandates.

Related #

• MRV Data Lineage & Provenance Tracking — the parent component this OpenLineage integration implements.
• MRV Architecture & Carbon Accounting Fundamentals — the wider pipeline stack these lineage events feed.
• Geospatial Coordinate Reference Systems (CRS) Alignment — the ingestion-stage projection discipline the spatial facet records.
• Carbon Credit Registry Data Integration — the downstream consumer that submits the traced figures.
• Step-by-Step GHG Protocol Scope 3 Geospatial Calculation — the per-record attribution logic these events make auditable.