Airflow Lineage Hooks with OpenLineage

Part of: Prefect vs Airflow for Geospatial Provenance

Emitting OpenLineage events from Airflow tasks lets you record the exact input and output geospatial datasets a step touched — along with the coordinate reference system each one carried — in a standardized, catalog-agnostic format, and you should reach for it whenever your compliance workflow needs tool-independent provenance rather than a bespoke payload. This how-to focuses on the practical mechanics of capturing input/output datasets and a CRS facet, and it assumes you have already weighed the orchestrator trade-offs in Prefect vs Airflow for Geospatial Provenance. For the general principle of attaching capture to lifecycle events, see Workflow Hooks in Python Pipelines.

The value of OpenLineage for geospatial work is that a reprojection step is not self-describing: the file path s3://out/scene_4326.tif hints at the output projection by convention, but nothing enforces that the file actually carries EPSG:4326, and a downstream catalog cannot reconcile datasets it only knows by name. By emitting a run event that names both datasets and pins each to a validated CRS facet, you turn a naming convention into an assertion that a machine can verify and an auditor can trust.

Prerequisites

  • Apache Airflow 2.7+ with the apache-airflow-providers-openlineage provider installed.
  • The openlineage-python client 1.x (pulled in by the provider) for constructing custom facets.
  • pyproj 3.6+ to resolve and validate the CRS you attach to each dataset facet.
  • An OpenLineage transport configured — an HTTP endpoint to a Marquez/OpenLineage backend, or a console / file transport for local verification. Set it via AIRFLOW__OPENLINEAGE__TRANSPORT or an openlineage.yml on the OPENLINEAGE_CONFIG path.
  • Environment: OPENLINEAGE_NAMESPACE set to your agency or pipeline namespace so events group correctly in the catalog.
OpenLineage event anatomy and where spatial detail attaches An event carries a run, a job, and input and output datasets; facets attach to each, with a custom spatial facet needed for CRS and extent. RunEvent — eventType START / COMPLETE / FAIL run runId (UUID) nominalTime facet job namespace + name sourceCode facet inputs / outputs namespace + name per dataset schema, dataSource facets CUSTOM spatial facet crs, extent, resolution, contentDigest No built-in facet carries CRS. A reprojection is invisible in stock OpenLineage — same schema in and out. Register a namespaced custom facet rather than smuggling CRS into a description string.

This is the single most important adaptation when applying OpenLineage to geospatial work. The standard’s built-in facets describe schema, row counts, column lineage and data quality — all of which are unchanged by a reprojection. Run a parcel layer from a state plane CRS to WGS 84 and stock OpenLineage records an input dataset and an output dataset with identical schemas and identical row counts, which is a faithful description of everything the standard knows how to see and a complete miss of what actually happened.

Define a custom facet under your own namespace carrying CRS, extent, resolution and content digest, and attach it to every input and output dataset. Custom facets are a first-class part of the spec rather than an escape hatch, so a conforming catalogue will store and return them; consumers that do not understand yours will ignore it rather than reject the event. Resist the temptation to encode CRS into the dataset name or a free-text description, which makes it unqueryable and breaks name stability across runs.

Implementation

The task below reprojects a raster and emits an OpenLineage run event that names the input and output datasets and attaches a custom CRS facet to each. Rather than relying solely on the provider’s automatic extraction, it constructs the event explicitly so the geospatial detail — which the generic extractors do not know about — is captured precisely.

from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Any

from airflow.decorators import task
from pyproj import CRS  # pyproj 3.6+, validates the EPSG code

from openlineage.client import OpenLineageClient
from openlineage.client.event_v2 import (
    Dataset,
    InputDataset,
    OutputDataset,
    Job,
    Run,
    RunEvent,
    RunState,
)
from openlineage.client.facet_v2 import BaseFacet


def crs_facet(epsg: str) -> dict[str, BaseFacet]:
    """Build a custom geospatial CRS facet, validated via pyproj."""
    crs = CRS.from_user_input(epsg)  # raises on an unknown/invalid code
    facet = BaseFacet()
    # attach fields directly; the emitted JSON keeps arbitrary custom keys
    facet.crs = epsg                      # e.g. "EPSG:4326"
    facet.crs_name = crs.name             # human-readable CRS name
    facet.is_projected = crs.is_projected
    facet.authority = ":".join(crs.to_authority() or ("EPSG", "0"))
    return {"geospatial_crs": facet}


def emit_reproject_lineage(
    input_uri: str,
    input_crs: str,
    output_uri: str,
    output_crs: str,
    namespace: str = "gis-agency",
) -> None:
    client = OpenLineageClient()  # transport read from OPENLINEAGE_CONFIG/env
    run = Run(runId=str(uuid.uuid4()))
    job = Job(namespace=namespace, name="reproject_raster")
    now = datetime.now(timezone.utc).isoformat()

    inputs = [InputDataset(
        namespace=namespace, name=input_uri, facets=crs_facet(input_crs),
    )]
    outputs = [OutputDataset(
        namespace=namespace, name=output_uri, facets=crs_facet(output_crs),
    )]

    for state in (RunState.START, RunState.COMPLETE):
        client.emit(RunEvent(
            eventType=state,
            eventTime=now,
            run=run,
            job=job,
            inputs=inputs if state == RunState.COMPLETE else [],
            outputs=outputs if state == RunState.COMPLETE else [],
            producer="https://gis-agency/airflow/reproject",
        ))


@task
def reproject_and_emit(input_uri: str, output_uri: str) -> str:
    # rasterio/pyproj reprojection would run here, EPSG:32633 -> EPSG:4326
    emit_reproject_lineage(
        input_uri=input_uri, input_crs="EPSG:32633",
        output_uri=output_uri, output_crs="EPSG:4326",
    )
    return output_uri

The key line is crs_facet(...), which validates the EPSG code through pyproj before it ever reaches the event — a wrong or missing CRS fails loudly at emit time rather than silently producing a provenance record that lies about the projection. Emitting both START and COMPLETE events, with datasets attached to the completion, mirrors the OpenLineage run lifecycle that catalogs expect: the START event opens the run and records the job, while the COMPLETE event carries the resolved inputs and outputs once the task knows exactly which files it read and wrote. Sharing a single runId across both events is what lets the catalog stitch them into one logical run rather than two orphaned records.

Two design choices in this task deserve emphasis. First, the datasets are attached to the COMPLETE event rather than the START event because, in a real reprojection, the output URI may only be finalized after processing — deferring dataset resolution avoids emitting a placeholder that later diverges from reality. Second, the facet is attached to both the input and the output dataset, not just one. A lineage record that states only the output CRS loses the very fact a reprojection audit cares about: that the transformation moved data from EPSG:32633 to EPSG:4326. Capturing the source projection alongside the target is what makes the event a defensible record of what the step actually did.

Avoiding duplicate runs in the catalogue The provider emits automatically for each task instance; an explicit emit with a fresh run id creates a second, unrelated run for the same work. One task instance reproject_parcels provider auto-event runId = A your explicit emit runId = B ← the bug catalogue shows two runs, one real Two fixes, pick one: reuse the provider's runId and emit only additional facets… …or disable automatic emission for those tasks and own the whole event. Doing neither yields inflated run counts and impact analysis that double-counts every derivation.

Reusing the provider’s run identifier is usually the better of the two, because it keeps the operational metadata Airflow contributes for free — try number, execution date, task duration — while letting you attach the spatial facet the provider knows nothing about. Emitting a second COMPLETE event against the same runId with only the extra facets is valid; the catalogue merges them.

Taking full ownership is the right call only when the provider’s automatic events are actively wrong for your DAG, which happens most often with dynamically mapped tasks where the provider’s inferred dataset names do not match what the task actually wrote. In that case turn the provider off for those tasks specifically rather than globally, or you lose correct automatic capture everywhere else.

Verification

Point the transport at a file or the console and inspect the emitted JSON. With a file transport, each event is one JSON object per line; confirm the CRS facet rode along on both datasets:

export OPENLINEAGE_CONFIG=./openlineage.yml   # transport: type: file, log: ./ol.jsonl
python -c "from dag_module import reproject_and_emit; \
    reproject_and_emit.function('s3://raw/scene.tif', 's3://out/scene_4326.tif')"
tail -n 1 ./ol.jsonl | python -m json.tool

A correct COMPLETE event contains an inputs array whose dataset carries facets.geospatial_crs.crs = "EPSG:32633" and an outputs array whose dataset shows "EPSG:4326". If the facets object is empty, the facet was not attached; if the whole event is missing, the transport is misconfigured — check the next section. You can also assert programmatically that both dataset facets resolve to valid authorities:

import json
with open("ol.jsonl") as fh:
    event = json.loads(fh.readlines()[-1])
assert event["eventType"] == "COMPLETE"
assert event["outputs"][0]["facets"]["geospatial_crs"]["crs"] == "EPSG:4326"

Namespace and Name Stability

Stable dataset naming versus a node per run Embedding a run timestamp in the dataset name creates a fresh node every execution; a stable logical name with the partition in a facet keeps one node with many versions. UNSTABLE — timestamp in the name parcels_20250601.gpkg parcels_20250602.gpkg parcels_20250603.gpkg … Three unrelated nodes. No lineage connects yesterday's parcels to today's. STABLE — logical name, partition as a facet gis://county/parcels one node, many versions facet: {partition: "2025-06-03", digest: "a41f…"} the date is data about the run, not part of the identity

Backfills make this concrete. Re-running a DAG for ninety historical dates against timestamped dataset names produces ninety disconnected nodes and no answer to “what does the parcels layer depend on”. With a stable logical name, the same backfill produces ninety runs against one dataset, and the partition facet distinguishes them — which is exactly the shape a catalogue’s impact analysis expects.

The namespace half deserves equal care. It should identify the storage system rather than the environment, so gis://county rather than gis://county-staging, with environment expressed as a job namespace instead. Encoding the environment into the dataset namespace means a dataset promoted from staging to production appears as two unrelated datasets, and the promotion — often the most audit-relevant step — has no edge representing it.

Gotchas & edge cases

  • Facet schema key naming. Custom facets must carry a _producer and _schemaURL in strict OpenLineage validators; some catalogs reject a facet lacking them. If your backend enforces the schema, subclass the facet with those fields set rather than attaching bare attributes, or the event will be dropped silently at ingestion.
  • Transport configuration precedence. The provider reads transport config from openlineage.yml, the OPENLINEAGE_* environment variables, and Airflow’s [openlineage] config section — in that resolution order. Mixed sources are the most common reason events “disappear”: a stray AIRFLOW__OPENLINEAGE__DISABLED=true or a wrong URL will suppress emission with no task-level error. Verify with a console transport first.
  • CRS on compound or custom projections. CRS.from_user_input accepts WKT and PROJ strings, not only EPSG codes, but a locally-defined datum without an authority yields an empty to_authority(). Guard for that case (as the fallback tuple above does) so the facet still records the WKT rather than emitting a null authority that downstream reconciliation cannot match.
  • Dataset naming stability across backfills. OpenLineage identifies a dataset by its namespace plus name, so the URI you pass must be deterministic. If your reprojection writes to a run-date-partitioned path, include the logical date in the name consistently — but never fold a random run ID or timestamp into it, or every backfill re-run mints a new dataset node and the lineage graph fragments into unconnected islands the catalog cannot collapse.
  • Listener vs. explicit emission overlap. With the OpenLineage provider enabled, Airflow already emits automatic events for each task instance. Explicitly emitting your own run event on top of that can produce duplicate runs in the catalog unless you either reuse the provider’s run ID or disable automatic extraction for the task. Decide on one path — provider-managed or hand-emitted — rather than running both blind, and confirm in the catalog that each task produces exactly one run.