How to integrate Apache Airflow with OpenLineage for end-to-end traceability

How to integrate Apache Airflow with OpenLineage for end-to-end traceability By the end of this walkthrough, every DAG run in your Airflow instance emits structured lineage events that name the exact tables each task read and wrote, and you can open a graph and answer "which upstream job produced this number" without grepping a single scheduler log. That is the whole promise. No manual documentation, no lineage spreadsheet that goes stale in three weeks. The orchestrator reports what it actually did, while it is doing it. The setup itself is short. What follows is ordered to surface the three failures that usually show up first, before they cost you an afternoon of guessing. Prerequisites and versions - Apache Airflow 2.11.0 or later, or any Airflow 3.x release. That is the minimum version supported by the current provider distribution. - Python 3.9 to 3.12. - Docker, to run a lineage backend locally. - A Postgres connection in Airflow ( postgres_default ) if you want to reproduce the SQL example exactly. The two packages that matter: the Airflow OpenLineage provider extracts Airflow metadata and turns it into events, and openlineage-python transmits them. The client can be upgraded independently of the provider, which is useful when you need a transport fix without touching your Airflow version. Step 1: the event model, before you install anything OpenLineage has three objects and one extension mechanism. Skipping this part is why most first integrations produce an empty graph. - Job: something that runs. Your DAG is a job, and each task is also a job. - Run: one execution of a job, with a unique run ID. - Dataset: something read or written. Identified by a namespace and aname . - Facet: an atomic block of metadata attached to any of the above. Schema, SQL text, column-level lineage, run state, and your own custom fields all arrive as facets, and the OpenLineage specification lists the standard ones. Events fire on state transitions: START , RUNNING , COMPLETE , FAIL , ABORT , OTHER . Lineage is reconstructed downstream by joining datasets across runs, which means dataset identity is the thing that makes or breaks the graph. More on that later, because it is the most common source of a graph with nodes and no edges. Step 2: install the provider pip install apache-airflow-providers-openlineage Official Airflow Docker images may already ship it. Check before adding it to your requirements file: airflow providers list | grep openlineage Nothing is emitted yet. The provider stays silent until it knows where to send events. Step 3: point it at a transport Start with the console transport. It writes events to the task logs, costs nothing to run, and tells you immediately whether extraction works at all. export AIRFLOW__OPENLINEAGE__TRANSPORT='{"type": "console"}' Once you see events in the logs, move to a real backend. Marquez is one option here, and any OpenLineage-compatible backend works. It is the reference implementation of the standard, which makes it the fastest way to get a lineage UI running: git clone https://github.com/MarquezProject/marquez cd marquez ./docker/up.sh The API listens on port 5000, the admin interface on 5001, and the web UI on 3000. On macOS, port 5000 is reserved by the operating system, so run ./docker/up.sh --api-port 9000 and adjust the URL below. Now switch the transport: export AIRFLOW__OPENLINEAGE__TRANSPORT='{"type": "http", "url": "http://localhost:5000", "endpoint": "api/v1/lineage"}' export AIRFLOW__OPENLINEAGE__NAMESPACE='airflow-local' The same thing in airflow.cfg : [openlineage] transport = {"type": "http", "url": "http://localhost:5000", "endpoint": "api/v1/lineage"} namespace = airflow-local disabled = False Set the namespace deliberately. It logically separates producers, so a staging Airflow and a production Airflow do not merge into one graph and lie to you. If you leave it unset, everything lands in default . For anything beyond local, do not put credentials in airflow.cfg . The provider accepts a Generic Airflow connection ID holding the transport config, including auth, in the connection extra. Step 4: run a DAG that actually produces lineage SQL operators are the best place to start, because the provider parses the query and derives inputs, outputs, and column-level relationships without you writing anything: # dags/openlineage_demo.py from datetime import datetime from airflow import DAG from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator with DAG( dag_id="openlineage_demo", start_date=datetime(2026, 1, 1), schedule="@daily", catchup=False, ) as dag: build_daily_orders = SQLExecuteQueryOperator( task_id="build_daily_orders", conn_id="postgres_default", sql=""" CREATE TABLE IF NOT EXISTS analytics.daily_orders AS SELECT o.order_date, c.region, COUNT(*) AS order_count, SUM(o.amount) AS revenue FROM raw.orders o JOIN raw.customers c ON c.customer_id = o.customer_id GROUP BY o.order_date, c.region; """, ) Trigger it. The resulting event should list raw.orders and raw.customers as inputs and analytics.daily_orders as the output, with a column-level facet mapping revenue back to o.amount . Step 5: verify it worked Three checks, in this order. Look at the graph. Open http://localhost:3000 , find the namespace you configured, and confirm the two source tables connect to the output table. Check which tasks are even reporting. This is the diagnostic that is easiest to miss. The DagRun START event carries an AirflowJobFacet listing every task in the DAG, each with an emits_ol_events boolean. That tells you ahead of time which operators will stay invisible, instead of leaving you to guess why half the graph is missing. Understand the silence. An EmptyOperator emits nothing by default, because Airflow does not schedule it the way it schedules real work. Add an on_execute or on_success callback, or a task outlet, if you need it represented. When task-level detail does not matter, the DagRun COMPLETE event carries an AirflowStateRunFacet with the state of every task in the run. Step 6: cover the operators that report nothing Automatic extraction covers SQL operators and many provider operators. Your own operators report nothing until you tell them what they touch. For operators you own, implement the OpenLineage methods directly: from airflow.models import BaseOperator class S3ToWarehouseOperator(BaseOperator): def init(self, *, source_bucket, source_key, target_table, **kwargs): super().init(**kwargs) self.source_bucket = source_bucket self.source_key = source_key self.target_table = target_table def execute(self, context): ... # your copy logic def get_openlineage_facets_on_complete(self, task_instance): # import locally: top-level Airflow imports here can be cyclical # and make extraction fail silently from airflow.providers.common.compat.openlineage.facet import Dataset from airflow.providers.openlineage.extractors import OperatorLineage return OperatorLineage( inputs=[ Dataset(namespace=f"s3://{self.source_bucket}", name=self.source_key) ], outputs=[ Dataset(namespace="postgres://warehouse:5432", name=self.target_table) ], ) Rules worth internalizing: - You must implement at least one of get_openlineage_facets_on_start() orget_openlineage_facets_on_complete(ti) . Ifon_complete is missing, the provider falls back toon_start . There is alsoget_openlineage_facets_on_failure(ti) , which by default reuses theon_complete logic. - Prefer on_complete whenever the real dataset names are only resolved duringexecute . Reporting a wildcard path on start and never correcting it produces a confident, wrong graph. - Import OpenLineage objects inside the method, never at module level. The listener is instantiated when the worker starts, so a top-level Airflow import can become circular and kill extraction without an obvious error. For third-party operators you cannot modify, write a custom extractor and register it: export AIRFLOW__OPENLINEAGE__EXTRACTORS='plugins.extractors.MyCustomExtractor' Step 7: attach your own context Since provider version 1.10.0, you can inject arbitrary run facets without touching operator code. Write a function that accepts the task instance and returns a facet dictionary, then register the import paths, separated by semicolons: export AIRFLOW__OPENLINEAGE__CUSTOM_RUN_FACETS='plugins.ol_facets.ownership_facet' This is how you get team, cost center, or change-ticket ID onto every event, which turns "who owns this broken pipeline" from a Slack thread into a filter. What breaks in production Four settings and one habit account for most of the pain. - include_full_task_info : tempting, and expensive. With it on, all serializable task parameters go into the event. Depending on what you pass to your tasks, single events can reach megabytes. - execution_timeout : cap how long extraction may run so a slow lineage call never becomes a pipeline incident. - dag_state_change_process_pool_size : processes the scheduler uses to handle DAG state changes asynchronously. Worth tuning on busy instances. - emission_policy : the current way to control what gets emitted. The olderselective_enable anddisable_source_code flags are deprecated in its favor. - Naming discipline: lineage joins on dataset identity. If one job writes analytics.daily_orders and another readsANALYTICS.DAILY_ORDERS , you get two nodes and no edge. Fix the convention (database.schema.table , environment-scoped namespaces) before you scale, because rewriting identities after the fact means reprocessing history. Where lineage stops being the answer What you have now is operational lineage: what ran, what it read, what it wrote, and whether it failed. That is enough to trace an incident backward and to run impact analysis before a schema change. It is not enough to answer who owns a dataset, whether it is certified, what "active customer" means in business terms, or where PII flows. Those live in a metadata platform, and OpenLineage events are the input to it rather than a repla