Picture this: it's 9 AM on a Tuesday and your head of analytics sends a Slack message that the weekly revenue dashboard is showing numbers that are 12% lower than last week with no obvious business explanation. You start pulling threads. The dbt model that powers the dashboard runs on top of a Snowflake table. That table is populated by an Airflow DAG that reads from S3. That S3 bucket gets written to by a Fivetran connector that pulls from a SaaS CRM. Somewhere in that chain, something broke or changed — but you have no automated way to find where. You are doing archaeology with SQL and Git blame.
This is the lineage problem, and it is not a minor inconvenience. As modern data stacks grow in sophistication — spanning ingestion tools, orchestration layers, transformation frameworks, semantic layers, and BI platforms — the gap between "we have logs" and "we understand data flow" becomes a serious operational liability. Debugging data quality issues, demonstrating regulatory compliance, performing impact analysis before schema changes, and building trust with business stakeholders all depend on your ability to answer a deceptively simple question: where did this data come from, and where does it go?
By the end of this lesson, you will have built a working multi-hop lineage tracking system using OpenLineage and Marquez that spans the full data stack — from raw ingestion events through transformation layers to final consumption surfaces. You will understand not just how to wire the plumbing, but why the architecture is designed the way it is, where it breaks down at scale, and how to extend it to meet enterprise requirements.
What you'll learn:
You should be comfortable with:
You do not need prior experience with OpenLineage or Marquez, though skimming the OpenLineage specification at openlineage.io before this lesson will help.
Before you touch any tooling, you need to understand what OpenLineage actually is, because most practitioners get this wrong. OpenLineage is not a tool — it is a specification. Specifically, it is an open standard for how data systems emit lineage events. Marquez is one implementation of a server that consumes those events. This distinction matters enormously for architectural decisions.
Every piece of lineage in the OpenLineage world is expressed as a RunEvent. A RunEvent captures a moment in time during the execution of a Job, which transforms zero or more input Datasets into zero or more output Datasets. Here is what that looks like in raw JSON:
{
"eventType": "START",
"eventTime": "2024-01-15T09:00:00.000Z",
"run": {
"runId": "d4b4c3e2-1234-5678-abcd-ef0123456789",
"facets": {
"nominalTime": {
"_producer": "https://github.com/OpenLineage/OpenLineage/tree/main/integration/airflow",
"_schemaURL": "https://openlineage.io/spec/facets/1-0-0/NominalTimeRunFacet.json",
"nominalStartTime": "2024-01-15T09:00:00.000Z"
}
}
},
"job": {
"namespace": "analytics_pipeline",
"name": "load_crm_contacts_to_s3",
"facets": {
"documentation": {
"_producer": "https://airflow.apache.org",
"_schemaURL": "https://openlineage.io/spec/facets/1-0-0/DocumentationJobFacet.json",
"description": "Extracts contacts from Salesforce and lands them in S3"
}
}
},
"inputs": [],
"outputs": [
{
"namespace": "s3://raw-data-bucket",
"name": "crm/contacts/2024/01/15",
"facets": {
"schema": {
"_producer": "https://airflow.apache.org",
"_schemaURL": "https://openlineage.io/spec/facets/1-0-0/SchemaDatasetFacet.json",
"fields": [
{"name": "contact_id", "type": "STRING"},
{"name": "email", "type": "STRING"},
{"name": "account_id", "type": "STRING"},
{"name": "created_at", "type": "TIMESTAMP"}
]
}
}
}
]
}
Notice the structure. Three top-level concepts: the run (this specific execution instance, identified by a UUID), the job (the reusable definition of the work, identified by namespace + name), and the datasets (inputs and outputs with their own namespace + name identifiers).
The eventType field is crucial for multi-hop tracking. Valid values are START, RUNNING, COMPLETE, FAIL, and ABORT. A lineage server that receives these events in order can reconstruct not just the graph topology, but the execution timeline — which matters when you are debugging a 3 AM pipeline failure.
The namespace + name combination is the universal identifier for a dataset across all systems. When your Airflow job writes to s3://raw-data-bucket/crm/contacts/2024/01/15 and your Spark job reads from that same path, they must emit the same namespace (s3://raw-data-bucket) and name (crm/contacts/2024/01/15) for the lineage graph to connect. If one system emits s3://raw-data-bucket/crm/contacts/2024/01/15 and another emits s3://raw-data-bucket/crm/contacts/2024/01/15/ (trailing slash), they will appear as different datasets in Marquez and the lineage chain breaks.
This is the most common source of lineage graph fragmentation in production systems. Establish namespace conventions before you instrument anything else. Write them down. Put them in your data engineering handbook.
Facets are the extension point in OpenLineage. The core spec defines the run/job/dataset structure; facets let you attach typed, versioned metadata to any of those objects. The schema facet shown above is one example. Others include:
columnLineage facet on output datasets: maps output columns back to the input columns they were derived from — this is the key to column-level lineagedataQualityMetrics facet: carries row counts, null counts, and custom assertionssqlJob facet on jobs: carries the actual SQL text that was executedsourceCodeLocation facet: links a job to its Git repository and commit SHACustom facets are legal and encouraged. If you want to attach your data contract version, your team ownership metadata, or your SLO tier to every lineage event, you define a custom facet with its own JSON Schema URL and emit it. Marquez will store it and make it queryable.
Marquez is your lineage backend for this lesson. It provides the OpenLineage-compatible HTTP API, a PostgreSQL-backed storage layer, and a web UI for graph visualization. Let's get it running.
# docker-compose.yml
version: "3.7"
services:
marquez:
image: marquezproject/marquez:0.45.0
ports:
- "5000:5000" # API
- "5001:5001" # Admin
environment:
- MARQUEZ_PORT=5000
- MARQUEZ_ADMIN_PORT=5001
- POSTGRES_HOST=marquez_db
- POSTGRES_PORT=5432
- POSTGRES_DB=marquez
- POSTGRES_USER=marquez
- POSTGRES_PASSWORD=marquez_secret
depends_on:
- marquez_db
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:5000/api/v1/namespaces"]
interval: 10s
timeout: 5s
retries: 5
marquez_web:
image: marquezproject/marquez-web:0.45.0
ports:
- "3000:3000"
environment:
- MARQUEZ_HOST=marquez
- MARQUEZ_PORT=5000
depends_on:
- marquez
marquez_db:
image: postgres:14
environment:
- POSTGRES_USER=marquez
- POSTGRES_PASSWORD=marquez_secret
- POSTGRES_DB=marquez
volumes:
- marquez_db_data:/var/lib/postgresql/data
volumes:
marquez_db_data:
docker-compose up -d
# Wait for health check to pass
curl http://localhost:5000/api/v1/namespaces
Once you see an empty namespaces list in the response, you are ready to start emitting events.
Tip: In production, Marquez sits behind a load balancer and talks to a managed PostgreSQL instance (RDS, Cloud SQL, etc.). The stateless API tier scales horizontally; the bottleneck is always the database. Keep that in mind for high-volume event streams.
The Apache Airflow integration is the most mature in the OpenLineage ecosystem. When you install apache-airflow-providers-openlineage, it hooks into Airflow's listener interface and emits RunEvent payloads for every task execution automatically — without you modifying your DAG code.
pip install apache-airflow-providers-openlineage==1.7.0
Configure the transport in your Airflow environment (environment variables or airflow.cfg):
# Environment variables (preferred for containerized deployments)
export OPENLINEAGE_URL=http://localhost:5000
export OPENLINEAGE_NAMESPACE=production_pipelines
export OPENLINEAGE_API_KEY= # leave empty for local Marquez without auth
Or in airflow.cfg:
[openlineage]
transport = {"type": "http", "url": "http://localhost:5000", "endpoint": "api/v1/lineage"}
namespace = production_pipelines
Let's build a DAG that simulates the CRM ingestion scenario from our opening story. This pipeline extracts data from a source system and lands it in S3.
# dags/crm_contacts_ingestion.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
import json
import uuid
default_args = {
"owner": "data_engineering",
"depends_on_past": False,
"retries": 2,
"retry_delay": timedelta(minutes=5),
"email_on_failure": True,
"email": ["data-eng-alerts@company.com"],
}
def extract_crm_contacts(execution_date: datetime, **context):
"""
In production this would call the Salesforce REST API.
We simulate it here with realistic data structure.
"""
contacts = [
{
"contact_id": str(uuid.uuid4()),
"email": f"contact_{i}@example.com",
"account_id": f"ACC{i:06d}",
"first_name": "Jane",
"last_name": "Smith",
"created_at": execution_date.isoformat(),
"is_active": True,
"annual_revenue_usd": 50000 + (i * 1000),
}
for i in range(100)
]
# Push to XCom for downstream task
context["ti"].xcom_push(key="contact_count", value=len(contacts))
return contacts
def load_contacts_to_s3(execution_date: datetime, **context):
"""Lands extracted contacts as newline-delimited JSON in S3."""
contacts = context["ti"].xcom_pull(
task_ids="extract_crm_contacts", key="return_value"
)
s3_hook = S3Hook(aws_conn_id="aws_default")
date_partition = execution_date.strftime("%Y/%m/%d")
s3_key = f"crm/contacts/{date_partition}/contacts.jsonl"
payload = "\n".join(json.dumps(c) for c in contacts)
s3_hook.load_string(
string_data=payload,
key=s3_key,
bucket_name="raw-data-lake",
replace=True,
)
return f"s3://raw-data-lake/{s3_key}"
with DAG(
dag_id="crm_contacts_ingestion",
default_args=default_args,
description="Ingest CRM contacts from Salesforce to S3 raw layer",
schedule_interval="0 2 * * *", # 2 AM daily
start_date=datetime(2024, 1, 1),
catchup=False,
tags=["ingestion", "crm", "contacts"],
) as dag:
extract_task = PythonOperator(
task_id="extract_crm_contacts",
python_callable=extract_crm_contacts,
op_kwargs={"execution_date": "{{ execution_date }}"},
)
load_task = PythonOperator(
task_id="load_contacts_to_s3",
python_callable=load_contacts_to_s3,
op_kwargs={"execution_date": "{{ execution_date }}"},
)
extract_task >> load_task
With the OpenLineage provider configured, Airflow will automatically emit START, COMPLETE (or FAIL) events for each task. But the automatic instrumentation has a limitation: it does not know what datasets your Python tasks read and write, because that logic lives inside your function.
To make the lineage graph connect across hops, you need to annotate your tasks with explicit input and output datasets. The OpenLineage Airflow provider supports this through the inlets and outlets mechanism:
from airflow.lineage.entities import File
from openlineage.airflow.extractors import TaskMetadata
# Alternative: use the OpenLineage Python client directly in the operator
from openlineage.client import OpenLineageClient
from openlineage.client.run import RunEvent, RunState, Run, Job, Dataset
from openlineage.client.facet import SchemaDatasetFacet, SchemaField, DataQualityMetricsInputDatasetFacet
def load_contacts_to_s3_with_lineage(execution_date: datetime, **context):
"""Enhanced version with explicit lineage emission."""
contacts = context["ti"].xcom_pull(
task_ids="extract_crm_contacts", key="return_value"
)
# ... actual S3 write logic ...
date_partition = execution_date.strftime("%Y/%m/%d")
s3_key = f"crm/contacts/{date_partition}/contacts.jsonl"
# Emit a custom lineage event with schema and quality metrics
client = OpenLineageClient.from_environment()
run_id = str(uuid.uuid4())
output_schema = SchemaDatasetFacet(
fields=[
SchemaField(name="contact_id", type="STRING"),
SchemaField(name="email", type="STRING"),
SchemaField(name="account_id", type="STRING"),
SchemaField(name="first_name", type="STRING"),
SchemaField(name="last_name", type="STRING"),
SchemaField(name="created_at", type="TIMESTAMP"),
SchemaField(name="is_active", type="BOOLEAN"),
SchemaField(name="annual_revenue_usd", type="DOUBLE"),
]
)
client.emit(
RunEvent(
eventType=RunState.COMPLETE,
eventTime=datetime.utcnow().isoformat() + "Z",
run=Run(runId=run_id),
job=Job(
namespace="production_pipelines",
name="crm_contacts_ingestion.load_contacts_to_s3"
),
inputs=[],
outputs=[
Dataset(
namespace="s3://raw-data-lake",
name=f"crm/contacts/{date_partition}/contacts.jsonl",
facets={"schema": output_schema}
)
]
)
)
Warning: If you emit custom lineage events from within your tasks in addition to the automatic provider events, you will get duplicate job runs in Marquez. Use either the automatic extraction or custom emission, not both for the same task. The exception is when you need the schema or data quality facets, which the automatic extractor cannot infer.
dbt has first-class OpenLineage support through the openlineage-dbt package, which works by reading dbt's manifest.json and run_results.json artifacts and translating them into OpenLineage events. This is a fundamentally different approach from the Airflow integration — it is post-hoc rather than real-time.
pip install openlineage-dbt
After a dbt run, emit lineage events from the generated artifacts:
# Run dbt as normal
dbt run --profiles-dir . --target prod
# Emit lineage to Marquez from the artifacts
dbt-ol send-events \
--producer https://github.com/OpenLineage/OpenLineage/tree/main/integration/dbt \
--target prod \
--project-dir . \
--profiles-dir .
Or integrate it into your Airflow orchestration:
from airflow.operators.bash import BashOperator
dbt_run = BashOperator(
task_id="dbt_run_contacts_marts",
bash_command="""
cd /opt/dbt/analytics && \
dbt run --select marts.contacts+ --target prod && \
dbt-ol send-events --target prod --project-dir . --profiles-dir .
""",
env={
"OPENLINEAGE_URL": "http://marquez:5000",
"OPENLINEAGE_NAMESPACE": "production_pipelines",
}
)
When openlineage-dbt processes your manifest, it translates dbt's native lineage model (which tracks model-to-model dependencies) into OpenLineage events. For a model like this:
-- models/marts/contacts/dim_contacts.sql
{{
config(
materialized='table',
schema='marts',
tags=['contacts', 'daily']
)
}}
with source_contacts as (
select * from {{ ref('stg_crm_contacts') }}
),
enriched_accounts as (
select * from {{ ref('stg_salesforce_accounts') }}
),
final as (
select
c.contact_id,
c.email,
c.first_name,
c.last_name,
c.created_at,
c.is_active,
a.account_name,
a.industry,
a.annual_revenue_usd,
-- Derived field: segment based on revenue
case
when a.annual_revenue_usd >= 1000000 then 'enterprise'
when a.annual_revenue_usd >= 100000 then 'mid_market'
else 'smb'
end as customer_segment
from source_contacts c
left join enriched_accounts a using (account_id)
)
select * from final
The emitted lineage event will have:
dbt.dim_contacts in the production_pipelines namespacesnowflake://company.snowflakecomputing.com/analytics/staging/stg_crm_contacts and snowflake://company.snowflakecomputing.com/analytics/staging/stg_salesforce_accountssnowflake://company.snowflakecomputing.com/analytics/marts/dim_contactsThis is the hop that connects S3 ingestion data → Snowflake staging tables → Snowflake mart tables. The staging models (stg_crm_contacts) are where the raw S3 data gets loaded by your data warehouse loader (Fivetran, Airbyte, or a custom ELT job). The lineage chain only fully closes if the S3-to-Snowflake loading step also emits events with matching dataset identifiers.
The columnLineage facet is where lineage goes from "nice to have" to operationally essential. With column-level lineage, you can answer: "If I add a NOT NULL constraint to stg_crm_contacts.email, which downstream columns will break?"
As of OpenLineage 1.9+, dbt emits column-level lineage for simple column references. For the model above, the contact_id column in dim_contacts traces directly back to contact_id in stg_crm_contacts. The customer_segment column traces back to annual_revenue_usd in stg_salesforce_accounts (with transformation type noted as TRANSFORM).
Complex expressions — window functions, CTEs with multiple aggregations, dynamic SQL — degrade the accuracy of automatic column-level lineage detection. For those cases, you can add explicit column lineage annotations in dbt's YAML config or emit custom facets.
When your transformation layer involves PySpark for heavy computation — joining billion-row fact tables, running ML feature engineering — you need the OpenLineage Spark integration, which works as a JVM agent injected into the Spark driver.
In your Spark submit command or your SparkConf:
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("revenue_fact_aggregation") \
.config(
"spark.jars.packages",
"io.openlineage:openlineage-spark_2.12:1.9.0"
) \
.config(
"spark.extraListeners",
"io.openlineage.spark.agent.OpenLineageSparkListener"
) \
.config("spark.openlineage.transport.type", "http") \
.config("spark.openlineage.transport.url", "http://marquez:5000") \
.config("spark.openlineage.transport.endpoint", "api/v1/lineage") \
.config("spark.openlineage.namespace", "production_pipelines") \
.config("spark.openlineage.appName", "revenue_fact_aggregation") \
.getOrCreate()
The Spark listener intercepts Spark's query plan events, reads the physical plan DAG, and extracts the actual dataset read/write operations — including the schemas. This means it captures lineage even from implicit reads (e.g., reading a partition from a Hive metastore table) without any code changes.
# This Spark job's lineage is captured entirely automatically
def build_revenue_facts(spark: SparkSession, run_date: str):
# Read from Snowflake (via Spark-Snowflake connector)
dim_contacts = spark.read \
.format("snowflake") \
.options(**snowflake_options) \
.option("dbtable", "analytics.marts.dim_contacts") \
.load()
# Read from S3 (Parquet format)
raw_transactions = spark.read \
.parquet(f"s3://raw-data-lake/transactions/date={run_date}/")
# Join and aggregate
revenue_facts = raw_transactions \
.join(dim_contacts, on="contact_id", how="left") \
.groupBy("customer_segment", "account_id") \
.agg(
F.sum("transaction_amount_usd").alias("total_revenue_usd"),
F.count("transaction_id").alias("transaction_count"),
F.avg("transaction_amount_usd").alias("avg_transaction_usd"),
)
# Write to Snowflake
revenue_facts.write \
.format("snowflake") \
.options(**snowflake_options) \
.option("dbtable", "analytics.marts.fct_revenue") \
.mode("overwrite") \
.save()
return revenue_facts.count()
The agent will emit START and COMPLETE events with the full input/output dataset list automatically. The physical plan parsing gives it column-level lineage for most standard Spark operations.
Warning: The Spark OpenLineage agent adds measurable overhead on large jobs — typically 2-5% on CPU-bound workloads, but potentially higher on jobs with complex query plans. Profile your jobs in staging before deploying to production on critical SLA paths.
Now that you have events flowing from all three layers, let's build the tooling to actually use the lineage data. The Marquez API is RESTful and reasonably well-documented, but there are some non-obvious behaviors that matter for production use.
Marquez stores the canonical representation of lineage as a graph where:
This matters because when you query the API, you are querying the current state of the graph — the latest known schema, the latest run status. Historical queries require working through the runs API.
The real power of lineage is in traversal. Let's build a Python utility that answers the archeological question from the introduction: given a dataset that looks wrong, find all upstream jobs and datasets.
import requests
from dataclasses import dataclass, field
from typing import Optional
from collections import deque
import json
MARQUEZ_BASE_URL = "http://localhost:5000/api/v1"
@dataclass
class LineageNode:
node_type: str # 'dataset' or 'job'
namespace: str
name: str
depth: int
parent: Optional[str] = None
last_run_state: Optional[str] = None
last_run_time: Optional[str] = None
schema_fields: list = field(default_factory=list)
def get_dataset_lineage(namespace: str, dataset_name: str, depth: int = 5) -> dict:
"""Fetch the lineage graph for a dataset from Marquez."""
encoded_namespace = requests.utils.quote(namespace, safe="")
encoded_name = requests.utils.quote(dataset_name, safe="")
url = f"{MARQUEZ_BASE_URL}/lineage"
params = {
"nodeId": f"dataset:{namespace}:{dataset_name}",
"depth": depth,
}
response = requests.get(url, params=params)
response.raise_for_status()
return response.json()
def get_latest_job_run(namespace: str, job_name: str) -> Optional[dict]:
"""Get the most recent run for a job."""
encoded_namespace = requests.utils.quote(namespace, safe="")
encoded_name = requests.utils.quote(job_name, safe="")
url = f"{MARQUEZ_BASE_URL}/namespaces/{encoded_namespace}/jobs/{encoded_name}/runs"
params = {"limit": 1, "offset": 0}
response = requests.get(url, params=params)
if response.status_code == 404:
return None
response.raise_for_status()
runs = response.json().get("runs", [])
return runs[0] if runs else None
def traverse_upstream_lineage(
namespace: str,
dataset_name: str,
max_depth: int = 10
) -> list[LineageNode]:
"""
BFS traversal of upstream lineage graph.
Returns a list of LineageNodes ordered by distance from the target dataset.
"""
lineage_graph = get_dataset_lineage(namespace, dataset_name, depth=max_depth)
visited = set()
result_nodes = []
# The Marquez lineage API returns a graph with nodes and edges
# We need to traverse it to find upstream nodes
graph_nodes = {node["id"]: node for node in lineage_graph.get("graph", [])}
# Build adjacency: for each node, what are its upstream nodes?
# An edge from A -> B means A is an input to B
# We want to traverse edges in reverse to find upstreams
for node in lineage_graph.get("graph", []):
node_id = node["id"]
node_data = node.get("data", {})
if node["type"] == "JOB":
latest_run = get_latest_job_run(
node_data.get("namespace", ""),
node_data.get("name", "")
)
lineage_node = LineageNode(
node_type="job",
namespace=node_data.get("namespace", ""),
name=node_data.get("name", ""),
depth=0, # Marquez doesn't return depth directly
last_run_state=latest_run.get("state") if latest_run else None,
last_run_time=latest_run.get("updatedAt") if latest_run else None,
)
result_nodes.append(lineage_node)
elif node["type"] == "DATASET":
facets = node_data.get("facets", {})
schema_facet = facets.get("schema", {})
fields = schema_facet.get("fields", [])
lineage_node = LineageNode(
node_type="dataset",
namespace=node_data.get("namespace", ""),
name=node_data.get("name", ""),
depth=0,
schema_fields=[f.get("name") for f in fields],
)
result_nodes.append(lineage_node)
return result_nodes
def find_failed_upstream_jobs(namespace: str, dataset_name: str) -> list[LineageNode]:
"""
Given a dataset that looks wrong, find all upstream jobs that have
recently failed. This is your 9 AM debugging tool.
"""
all_nodes = traverse_upstream_lineage(namespace, dataset_name)
failed_jobs = [
node for node in all_nodes
if node.node_type == "job" and node.last_run_state in ("FAILED", "ABORTED")
]
return failed_jobs
def print_lineage_report(namespace: str, dataset_name: str):
"""Generate a human-readable upstream lineage report."""
print(f"\n{'='*60}")
print(f"UPSTREAM LINEAGE REPORT")
print(f"Target Dataset: {namespace}/{dataset_name}")
print(f"{'='*60}\n")
nodes = traverse_upstream_lineage(namespace, dataset_name, max_depth=8)
job_nodes = [n for n in nodes if n.node_type == "job"]
dataset_nodes = [n for n in nodes if n.node_type == "dataset"]
print(f"Found {len(job_nodes)} upstream jobs and {len(dataset_nodes)} upstream datasets\n")
print("JOBS:")
for job in job_nodes:
status_emoji = {
"COMPLETED": "✅",
"FAILED": "❌",
"RUNNING": "🔄",
"ABORTED": "⚠️"
}.get(job.last_run_state, "❓")
print(f" {status_emoji} {job.namespace}/{job.name}")
if job.last_run_time:
print(f" Last run: {job.last_run_time} [{job.last_run_state}]")
print("\nDATASETS:")
for ds in dataset_nodes:
field_preview = ", ".join(ds.schema_fields[:5])
if len(ds.schema_fields) > 5:
field_preview += f" ... (+{len(ds.schema_fields)-5} more)"
print(f" 📊 {ds.namespace}/{ds.name}")
if field_preview:
print(f" Fields: {field_preview}")
if __name__ == "__main__":
# Investigate the revenue dashboard dataset
print_lineage_report(
namespace="snowflake://company.snowflakecomputing.com",
dataset_name="analytics/marts/fct_revenue"
)
# Find failed jobs
failed = find_failed_upstream_jobs(
namespace="snowflake://company.snowflakecomputing.com",
dataset_name="analytics/marts/fct_revenue"
)
if failed:
print(f"\n🚨 ALERT: Found {len(failed)} failed upstream jobs!")
for job in failed:
print(f" - {job.name} failed at {job.last_run_time}")
Upstream lineage tells you where problems come from. Downstream lineage tells you what breaks when you make a change. This is critical before schema changes, deprecations, or migrations.
def find_downstream_consumers(namespace: str, dataset_name: str) -> dict:
"""
Before deprecating a dataset or changing its schema,
find everything that depends on it.
"""
lineage_graph = get_dataset_lineage(namespace, dataset_name, depth=5)
downstream_jobs = []
downstream_datasets = []
# In the Marquez lineage graph, we need to find nodes where
# our target dataset appears as an input
target_id = f"dataset:{namespace}:{dataset_name}"
for node in lineage_graph.get("graph", []):
if node["type"] == "JOB":
# Check if our dataset appears in this job's inputs
in_edges = node.get("inEdges", [])
if any(edge.get("origin") == target_id for edge in in_edges):
downstream_jobs.append(node["data"])
return {
"target": f"{namespace}/{dataset_name}",
"downstream_job_count": len(downstream_jobs),
"downstream_jobs": downstream_jobs,
"impact_assessment": "HIGH" if len(downstream_jobs) > 5 else "MEDIUM" if len(downstream_jobs) > 1 else "LOW"
}
What works on your laptop breaks in interesting ways at scale. Here is what you will actually encounter when you deploy this in a real organization.
OpenLineage events are delivered over HTTP with no ordering guarantees beyond best-effort. In a distributed system with many parallel pipeline runs, you will receive events out of order. Marquez handles this reasonably well — it uses the eventTime field for ordering when reconstructing run state — but there are edge cases.
If a COMPLETE event arrives before the corresponding START event (common when pipelines retry tasks), Marquez creates the run with whatever state it first sees. The subsequent START event is then ignored because the run already exists. This produces runs that show as COMPLETE with no start time in the UI. The fix is to ensure your transport is reliable — use an async event buffer (Kafka or SQS) between your emitters and Marquez rather than direct HTTP when event volume is high.
[Airflow Task] ──HTTP──► [Kafka Topic: lineage-events] ──► [Marquez Consumer] ──► [Marquez DB]
[Spark Job] ──HTTP──► [Kafka Topic: lineage-events] ──► [Marquez Consumer] ──► [Marquez DB]
[dbt Run] ──HTTP──► [Kafka Topic: lineage-events] ──► [Marquez Consumer] ──► [Marquez DB]
The openlineage-python client supports Kafka transport natively:
# openlineage.yml for Kafka transport
transport:
type: kafka
config:
bootstrap.servers: kafka-broker-1:9092,kafka-broker-2:9092
topic: openlineage.events
security.protocol: SASL_SSL
sasl.mechanism: PLAIN
sasl.username: ${KAFKA_USERNAME}
sasl.password: ${KAFKA_PASSWORD}
As your lineage graph grows to thousands of datasets, namespace collisions become a real problem. Two common failure modes:
Fragmented lineage from inconsistent naming: Your Airflow job emits s3://raw-data-lake/crm/contacts/2024/01/15/contacts.jsonl but your Spark job reads s3://raw-data-lake/crm/contacts/2024/01/15 (without the filename). These are different dataset identifiers — the graph breaks.
Over-granular partitions: If every daily S3 partition is a separate dataset, your lineage graph explodes with thousands of near-identical nodes. For S3-based data lakes, it is often better to use the partition root (crm/contacts) as the dataset name and express the partition as a facet. This requires custom instrumentation.
Build a LineageNamingConvention helper that all your teams use:
class LineageNamingConvention:
"""
Centralized dataset naming to prevent graph fragmentation.
All teams import this and use it when emitting custom lineage events.
"""
@staticmethod
def s3_dataset(bucket: str, prefix: str, include_partition: bool = False,
partition_date: str = None) -> tuple[str, str]:
"""Returns (namespace, name) for an S3 dataset."""
namespace = f"s3://{bucket}"
# Normalize: remove trailing slashes, lowercase
name = prefix.strip("/").lower()
if include_partition and partition_date:
name = f"{name}/date={partition_date}"
return namespace, name
@staticmethod
def snowflake_dataset(account: str, database: str, schema: str,
table: str) -> tuple[str, str]:
"""Returns (namespace, name) for a Snowflake table."""
namespace = f"snowflake://{account}.snowflakecomputing.com"
# Snowflake identifiers are case-insensitive; normalize to lowercase
name = f"{database.lower()}/{schema.lower()}/{table.lower()}"
return namespace, name
@staticmethod
def job_name(dag_id: str, task_id: str = None) -> str:
"""Consistent job naming across Airflow and custom emitters."""
if task_id:
return f"{dag_id}.{task_id}"
return dag_id
Lineage metadata can be sensitive. The graph of your data flows reveals your business processes, vendor relationships, and data architecture to anyone who can access the Marquez API. In production, you need:
dataQualityMetrics facet can include value distributions — scrub these in your instrumentation code before emitting.When your source schema changes — a new column added, a column renamed — the lineage graph needs to reflect this. OpenLineage handles this naturally because each RunEvent carries the current schema in the SchemaDatasetFacet. Marquez retains the latest schema on each dataset and the historical schemas through the runs.
The challenge is detecting breaking schema changes automatically. Here is a pattern that wraps schema comparison into your lineage emission:
def detect_schema_change(
client: OpenLineageClient,
namespace: str,
dataset_name: str,
new_fields: list[SchemaField]
) -> list[str]:
"""
Compare the schema about to be written against the
last known schema in Marquez. Returns list of breaking changes.
"""
encoded_ns = requests.utils.quote(namespace, safe="")
encoded_name = requests.utils.quote(dataset_name, safe="")
url = f"{MARQUEZ_BASE_URL}/namespaces/{encoded_ns}/datasets/{encoded_name}"
response = requests.get(url)
if response.status_code == 404:
return [] # New dataset, no breaking changes
response.raise_for_status()
current_dataset = response.json()
current_fields = {
f["name"]: f["type"]
for f in current_dataset.get("fields", [])
}
new_field_map = {f.name: f.type for f in new_fields}
breaking_changes = []
# Check for removed columns
for col_name in current_fields:
if col_name not in new_field_map:
breaking_changes.append(f"COLUMN_REMOVED: {col_name}")
# Check for type changes
for col_name, new_type in new_field_map.items():
if col_name in current_fields and current_fields[col_name] != new_type:
breaking_changes.append(
f"TYPE_CHANGED: {col_name} from {current_fields[col_name]} to {new_type}"
)
return breaking_changes
You are going to build the complete multi-hop lineage pipeline for a simplified e-commerce analytics stack. This exercise connects three hops: raw order ingestion → Snowflake staging → mart aggregation.
Setup: Start your Marquez instance with the docker-compose file from earlier.
Step 1: Write and emit a lineage event representing an S3 ingestion job that writes raw orders:
# exercise/step1_emit_ingestion_lineage.py
from openlineage.client import OpenLineageClient
from openlineage.client.run import RunEvent, RunState, Run, Job, Dataset
from openlineage.client.facet import (
SchemaDatasetFacet, SchemaField,
DataQualityMetricsInputDatasetFacet,
NominalTimeRunFacet,
)
import uuid
from datetime import datetime
client = OpenLineageClient.from_environment()
run_id = str(uuid.uuid4())
run_date = "2024-01-15"
# Step 1a: Emit START event
client.emit(RunEvent(
eventType=RunState.START,
eventTime=f"{run_date}T02:00:00.000Z",
run=Run(
runId=run_id,
facets={"nominalTime": NominalTimeRunFacet(
nominalStartTime=f"{run_date}T02:00:00.000Z"
)}
),
job=Job(namespace="ecommerce_pipelines", name="ingest_orders_to_s3"),
inputs=[],
outputs=[]
))
print(f"Emitted START event for run {run_id}")
# Simulate the actual ingestion work...
import time
time.sleep(2)
# Step 1b: Emit COMPLETE event with schema and quality metrics
order_schema = SchemaDatasetFacet(fields=[
SchemaField(name="order_id", type="STRING"),
SchemaField(name="customer_id", type="STRING"),
SchemaField(name="order_date", type="DATE"),
SchemaField(name="total_amount_usd", type="DOUBLE"),
SchemaField(name="status", type="STRING"),
SchemaField(name="item_count", type="INTEGER"),
])
client.emit(RunEvent(
eventType=RunState.COMPLETE,
eventTime=f"{run_date}T02:15:00.000Z",
run=Run(runId=run_id),
job=Job(namespace="ecommerce_pipelines", name="ingest_orders_to_s3"),
inputs=[],
outputs=[
Dataset(
namespace="s3://raw-data-lake",
name=f"orders/date={run_date}",
facets={"schema": order_schema}
)
]
))
print(f"Emitted COMPLETE event. Check Marquez at http://localhost:3000")
Step 2: Emit a lineage event for the Snowflake staging load that reads from S3:
# exercise/step2_emit_staging_lineage.py
# Your task: emit lineage for a job named "load_stg_orders" that:
# - Reads from: s3://raw-data-lake/orders/date=2024-01-15
# - Writes to: snowflake://company.snowflakecomputing.com as analytics/staging/stg_orders
# - Include the same schema fields plus a "_loaded_at" TIMESTAMP field
# Hint: the namespace for the input MUST match Step 1's output namespace exactly
# TODO: implement this
Step 3: Emit lineage for a dbt-style mart model:
# exercise/step3_emit_mart_lineage.py
# Your task: emit lineage for a job named "dbt.fct_daily_orders" that:
# - Reads from: snowflake analytics/staging/stg_orders
# - Reads from: snowflake analytics/staging/stg_customers
# - Writes to: snowflake analytics/marts/fct_daily_orders
# - Include a columnLineage facet showing total_revenue_usd comes from total_amount_usd
# TODO: implement this
Step 4: Query the lineage graph and produce a report:
# exercise/step4_query_lineage.py
# Use the traverse_upstream_lineage function from earlier to:
# 1. Find all upstream datasets of analytics/marts/fct_daily_orders
# 2. Print a lineage report
# 3. Simulate a job failure by re-emitting Step 2 with RunState.FAIL
# 4. Re-run find_failed_upstream_jobs and verify it detects the failure
# TODO: implement this
After completing all four steps, open the Marquez UI at http://localhost:3000 and navigate through the graph visually. You should see three connected hops: S3 → Snowflake staging → Snowflake mart.
Symptom: Your lineage graph shows disconnected islands — the ingestion job's output and the transformation job's input exist as separate dataset nodes instead of being connected.
Diagnosis: Compare the namespace and name values in the emitted events for the shared dataset. Use the Marquez API directly:
curl "http://localhost:5000/api/v1/namespaces/s3%3A%2F%2Fraw-data-lake/datasets" | python -m json.tool
Fix: Standardize namespace conventions in code (see the LineageNamingConvention class above) and audit existing events. You can soft-correct this in Marquez by using the /api/v1/namespaces/{ns}/datasets/{name} PATCH endpoint to merge datasets, though this is a manual operation.
Symptom: Sporadic gaps in lineage — some pipeline runs are missing from Marquez even though the jobs completed successfully.
Diagnosis: Check your Airflow worker logs for OpenLineage transport errors. The default HTTP transport in the Airflow provider uses synchronous calls — under heavy task concurrency, the HTTP timeout can cause lineage events to be dropped silently.
Fix: Configure the transport to use an async HTTP client, or move to the Kafka transport. Also check Marquez's API server logs for 429 Too Many Requests responses, which indicate you are hitting rate limits.
# In airflow.cfg, configure async transport
[openlineage]
transport = {
"type": "http",
"url": "http://marquez:5000",
"endpoint": "api/v1/lineage",
"timeout": 10,
"retry": 3,
"async": true
}
Symptom: Marquez shows jobs as COMPLETE for pipeline runs where the task was actually skipped (Airflow ShortCircuitOperator) or where dbt used a cached result.
Diagnosis: The Airflow OpenLineage provider emits events based on task state, but skipped tasks in Airflow emit COMPLETE events (because the task state is SKIPPED, not FAILED). dbt's --select with partial refreshes can produce similar issues.
Fix: For critical lineage accuracy, emit custom events for skip/cache conditions with a custom facet indicating the execution mode. This requires wrapping your operators.
Symptom: Marquez API becomes slow, PostgreSQL storage grows faster than expected, and event emission starts timing out.
Diagnosis: Check the size of your facets. If you are emitting dataQualityMetrics with value distributions or sqlJob facets with 10,000-character SQL queries, individual events can exceed 1 MB. Marquez stores these as JSONB in PostgreSQL, and large JSONB values are expensive.
Fix: Set size limits on facet payloads in your emission layer. For SQL facets, truncate at 4000 characters. For data quality metrics, emit summary statistics (min, max, mean, count) rather than full value distributions.
Symptom: Lineage is missing for dbt runs that happen in CI (pull request validation) but present for production runs.
Diagnosis: The dbt-ol send-events command is not part of your CI pipeline, only your production deployment.
Fix: Add lineage emission to CI pipeline as well, using a separate Marquez namespace (ci_preview vs production). This also lets you see lineage changes in PRs — a very useful code review tool.
You have covered substantial ground. Let's consolidate:
The OpenLineage protocol gives you a vendor-neutral event format for expressing data lineage — jobs, datasets, runs, and facets. The critical design insight is that dataset identity is fully determined by namespace + name, which means consistency in naming conventions is not a best practice but a correctness requirement.
Airflow, dbt, and Spark integrations give you automatic lineage capture at each layer of the stack with minimal code changes. The Airflow provider hooks into the listener interface, dbt-ol reads manifest artifacts post-run, and the Spark agent parses physical query plans. Each approach has trade-offs between coverage, accuracy, and performance overhead.
Marquez provides a production-ready storage and query layer for lineage events, with a REST API that supports both dataset/job metadata queries and graph traversal. For high-volume systems, move the transport layer to Kafka to decouple emission from ingestion.
Multi-hop lineage traversal is the operational payoff — programmatic graph traversal turns a reactive debugging tool into a proactive impact analysis system. The patterns shown here (upstream failure detection, downstream impact assessment, schema change detection) are the building blocks of automated data observability.
Column-level lineage at scale: Explore OpenLineage's ColumnLineage facet in depth and integrate it with your dbt models and Spark jobs. Column-level lineage is where the real analytical power lives.
Integrate with DataHub or Apache Atlas: Marquez is excellent for greenfield deployments, but many organizations already have a metadata catalog. Both DataHub and Apache Atlas can consume OpenLineage events, letting you unify lineage with the broader data catalog.
Build a lineage-driven alerting system: Connect your lineage traversal queries to your alerting infrastructure. When a job emits a FAIL event, automatically query downstream consumers and send targeted alerts to dataset owners rather than generic "pipeline failed" notifications.
Lineage for ML pipelines: OpenLineage's MLJob facets support model training runs — connecting your feature engineering pipelines to model versions to serving endpoints closes the lineage loop for ML systems.
Explore Great Expectations + OpenLineage: The combination of data quality assertions and lineage creates a powerful observability stack where quality gate failures are automatically attributed to their upstream source.
The data engineering ecosystem is moving rapidly toward treating lineage as infrastructure rather than an afterthought. Getting this foundation right — consistent namespacing, reliable event transport, programmatic traversal — is an investment that compounds as your data stack grows in complexity.
Learning Path: Modern Data Stack