Chapter 7

OpenLineage Standard

OpenLineage is an open standard for lineage metadata collection, designed to solve a critical problem: vendor lock-in and lack of interoperability in data lineage tools.

The Interoperability Problem

Before OpenLineage, each data tool maintained its own proprietary lineage format:

Result: Fragmented lineage with no unified view across the data stack.

OpenLineage Solution: A common specification that any tool can emit and any lineage backend can consume, enabling end-to-end lineage across heterogeneous systems.

OpenLineage Architecture

Core Components

  1. Event Specification: JSON schema for lineage events
  2. Client Libraries: SDKs for emitting events (Python, Java, etc.)
  3. Integrations: Pre-built emitters for popular tools
  4. Backend: System that collects and stores lineage
OpenLineage Flow
┌──────────────┐      ┌──────────────┐      ┌──────────────┐
│   Airflow    │─────▶│ OpenLineage  │─────▶│   Lineage    │
│   (emitter)  │      │    Event     │      │   Backend    │
└──────────────┘      └──────────────┘      └──────────────┘
                             │
┌──────────────┐             │               ┌──────────────┐
│     dbt      │─────────────┤              │  Marquez     │
│   (emitter)  │             │──────────────▶│  (consumer)  │
└──────────────┘             │               └──────────────┘
                             │
┌──────────────┐             │               ┌──────────────┐
│    Spark     │─────────────┘               │   Custom     │
│   (emitter)  │                             │   System     │
└──────────────┘                             └──────────────┘

OpenLineage Event Model

Event Structure

Every OpenLineage event contains:

Complete OpenLineage Event
{
  "eventType": "COMPLETE",
  "eventTime": "2025-12-26T10:30:15.123Z",
  "run": {
    "runId": "550e8400-e29b-41d4-a716-446655440000",
    "facets": {
      "nominalTime": {
        "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.0.0",
        "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/NominalTimeRunFacet.json",
        "nominalStartTime": "2025-12-26T10:00:00Z"
      }
    }
  },
  "job": {
    "namespace": "my-scheduler",
    "name": "etl.customer-data-pipeline",
    "facets": {
      "documentation": {
        "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.0.0",
        "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/DocumentationJobFacet.json",
        "description": "Daily customer data aggregation"
      },
      "sourceCode": {
        "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.0.0",
        "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/SourceCodeJobFacet.json",
        "language": "sql",
        "source": "SELECT customer_id, COUNT(*) FROM orders GROUP BY customer_id"
      }
    }
  },
  "inputs": [
    {
      "namespace": "postgres://warehouse.company.com:5432",
      "name": "public.orders",
      "facets": {
        "schema": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.0.0",
          "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/SchemaDatasetFacet.json",
          "fields": [
            {"name": "order_id", "type": "INTEGER"},
            {"name": "customer_id", "type": "INTEGER"},
            {"name": "amount", "type": "DECIMAL"}
          ]
        },
        "dataQuality": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.0.0",
          "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/DataQualityDatasetFacet.json",
          "rowCount": 15247,
          "bytes": 1048576,
          "columnMetrics": {
            "customer_id": {
              "nullCount": 0,
              "distinctCount": 1834
            }
          }
        }
      }
    }
  ],
  "outputs": [
    {
      "namespace": "postgres://warehouse.company.com:5432",
      "name": "public.customer_summary",
      "facets": {
        "schema": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.0.0",
          "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/SchemaDatasetFacet.json",
          "fields": [
            {"name": "customer_id", "type": "INTEGER"},
            {"name": "order_count", "type": "INTEGER"}
          ]
        },
        "outputStatistics": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.0.0",
          "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/OutputStatisticsOutputDatasetFacet.json",
          "rowCount": 1834,
          "size": 73360
        }
      }
    }
  ],
  "producer": "https://github.com/OpenLineage/OpenLineage/tree/1.0.0"
}

Facets: Extensible Metadata

Facets are the key to OpenLineage's flexibility. They allow adding custom metadata without breaking the core specification.

Standard Facets

Run Facets

Job Facets

Dataset Facets

Custom Facets

Organizations can define their own facets:

Custom Facet Example
{
  "customBusinessContext": {
    "_producer": "https://company.com/lineage-extension/1.0.0",
    "_schemaURL": "https://company.com/schemas/BusinessContextFacet.json",
    "businessDomain": "finance",
    "costCenter": "CC-1234",
    "dataClassification": "confidential",
    "retentionPeriod": "7_years",
    "regulatoryFrameworks": ["GDPR", "SOX"]
  }
}

Integration Examples

Python Client

Emitting OpenLineage Events in Python
from openlineage.client import OpenLineageClient
from openlineage.client.run import RunEvent, RunState, Run, Job
from openlineage.client.facet import SqlJobFacet, SchemaDatasetFacet, SchemaField
import uuid
from datetime import datetime

# Initialize client
client = OpenLineageClient(url="http://lineage-backend:5000")

# Create unique run ID
run_id = str(uuid.uuid4())

# START event
client.emit(
    RunEvent(
        eventType=RunState.START,
        eventTime=datetime.now().isoformat(),
        run=Run(runId=run_id),
        job=Job(
            namespace="my-data-pipeline",
            name="transform-customer-data",
            facets={
                "sql": SqlJobFacet(
                    query="SELECT * FROM raw_customers WHERE active = true"
                )
            }
        ),
        inputs=[
            Dataset(
                namespace="postgres://db.company.com:5432",
                name="public.raw_customers",
                facets={
                    "schema": SchemaDatasetFacet(
                        fields=[
                            SchemaField(name="id", type="INTEGER"),
                            SchemaField(name="email", type="VARCHAR"),
                            SchemaField(name="active", type="BOOLEAN")
                        ]
                    )
                }
            )
        ],
        outputs=[
            Dataset(
                namespace="postgres://db.company.com:5432",
                name="public.active_customers"
            )
        ]
    )
)

# ... perform transformation ...

# COMPLETE event
client.emit(
    RunEvent(
        eventType=RunState.COMPLETE,
        eventTime=datetime.now().isoformat(),
        run=Run(runId=run_id),
        job=Job(namespace="my-data-pipeline", name="transform-customer-data"),
        inputs=[...],
        outputs=[...]
    )
)

Airflow Integration

Airflow with OpenLineage
from airflow import DAG
from airflow.providers.openlineage.extractors import OperatorLineageExtractor

# OpenLineage automatically captures lineage for:
# - SQL operators
# - Spark operators
# - Python operators (with decorators)

dag = DAG('customer_pipeline', ...)

# SQL task - lineage extracted automatically
sql_task = PostgresOperator(
    task_id='aggregate_customers',
    sql='INSERT INTO customer_summary SELECT ...',
    postgres_conn_id='warehouse'
)

# Lineage event emitted automatically with:
# - Input tables parsed from SQL
# - Output tables parsed from SQL
# - Execution timestamps
# - Task metadata

OpenLineage Backends

Marquez

Reference implementation of OpenLineage backend:

Commercial Options

Custom Backend

Build your own lineage collector:

from flask import Flask, request
app = Flask(__name__)

@app.route('/api/v1/lineage', methods=['POST'])
def collect_lineage():
    event = request.json

    # Validate OpenLineage schema
    # Store in your database
    # Update lineage graph
    # Trigger impact analysis

    return {'status': 'received'}, 200

Benefits of OpenLineage

  1. Vendor Neutrality: Not locked into proprietary format
  2. Ecosystem Growth: More tools adding support
  3. Best Practices: Standardized approach to lineage
  4. Community Support: Active open-source community
  5. Future-Proof: Adapts as data stack evolves

Adoption Strategy

  1. Start Small: One pipeline or tool
  2. Choose Backend: Marquez or commercial option
  3. Instrument Jobs: Add OpenLineage emitters
  4. Validate Events: Ensure quality metadata
  5. Expand Coverage: Add more integrations
  6. Build on Top: Impact analysis, visualization, etc.
Key Takeaway: OpenLineage transforms lineage from a proprietary afterthought into a first-class, interoperable data product. It's the TCP/IP of data lineage.