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 a name.
• 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() or get_openlineage_facets_on_complete(ti). If on_complete is missing, the provider falls back to on_start. There is also get_openlineage_facets_on_failure(ti), which by default reuses the on_complete logic.
• Prefer on_complete whenever the real dataset names are only resolved during execute. 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 older selective_enable and disable_source_code flags are deprecated in its favor.
• Naming discipline: lineage joins on dataset identity. If one job writes analytics.daily_orders and another reads ANALYTICS.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 replacement for it. If that is the direction you are heading, this breakdown of data governance and end-to-end lineage with a metadata catalog covers how the collection layer and the governance layer divide the work.

The order matters, though. Emit first, catalog second. A catalog fed by hand ages exactly as fast as the documentation you were trying to replace.