The Complete Production SOP: How to Deploy and Run OpenLineage Operational Data Lineage & Governance in Autonomous AI Agents

---
name: "openlineage-metadata-data-lineage"
description: "Collect end-to-end dataset lineage and job run metadata across Airflow, Spark, and dbt."
metadata:
  author: "OpenLineage"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "configure openlineage airflow"
  - "collect spark data lineage"
  - "stream lineage to marquez"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# OpenLineage Operational Data Lineage & Governance

Integrates OpenLineage: configuring Airflow OpenLineage provider, Spark OpenLineage agents, extracting dataset inputs/outputs and column-level facets, and streaming lineage events to Marquez or Atlan.

transport:
  type: http
  url: http://localhost:5000
  endpoint: api/v1/lineage
  timeout: 5.0
namespace: production_warehouse
circuitBreaker:
  type: simple
  memoryThreshold: 85

spark-submit \
  --master local[*] \
  --packages io.openlineage:openlineage-spark_2.12:1.15.0 \
  --conf spark.extraListeners=io.openlineage.spark.agent.OpenLineageSparkListener \
  --conf spark.openlineage.transport.type=http \
  --conf spark.openlineage.transport.url=http://localhost:5000 \
  --conf spark.openlineage.namespace=production_warehouse \
  --conf spark.openlineage.appName=customer_aggregations \
  transform_customers.py

import json
from jsonschema import validate

def test_openlineage_run_event_structure():
    sample_event = {
        "eventType": "COMPLETE",
        "eventTime": "2026-03-30T10:00:00.000Z",
        "run": {"runId": "ea445b5c-22eb-457a-bee1-44be2ced59aa"},
        "job": {"namespace": "production_warehouse", "name": "customer_aggregations"},
        "inputs": [{"namespace": "postgres://db:5432", "name": "raw.customers"}],
        "outputs": [{"namespace": "snowflake://wh", "name": "analytics.dim_customers"}],
        "producer": "https://github.com/OpenLineage/OpenLineage/tree/1.15.0/client/python"
    }
    assert sample_event["eventType"] in ["START", "RUNNING", "COMPLETE", "FAIL", "ABORT"]
    assert len(sample_event["inputs"]) > 0
    assert sample_event["outputs"][0]["name"] == "analytics.dim_customers"