data-lineageUse this skill when asked about data lineage, OpenLineage, Marquez, DataHub, column-level lineage, impact analysis, data provenance, data dependency tracking, or lineage graph models. This skill enforces: OpenLineage integration for standardized lineage collection, Marquez or DataHub deployment for lineage storage and querying, column-level lineage with SQL…
Independent GitHub snapshot · Automated classification · Original documentation governs
Capture, store, query, and visualize end-to-end data lineage from source systems through transformations to dashboards, supporting impact analysis, root cause investigation, and data governance compliance.
Exact user phrases: "data lineage", "OpenLineage", "Marquez", "DataHub lineage", "column-level lineage", "impact analysis", "data provenance", "lineage graph", "data dependency", "producer consumer lineage", "field-level lineage".
Before activating, verify:
Lineage configuration with OpenLineage integration, Marquez deployment, column-level lineage SQL parser config, and impact analysis report.
# OpenLineage integration config
# Marquez deployment
# Lineage event emission
# SQL parser setup
# Lineage extraction queries
No preamble. No postamble. No explanations. No filler/hedging/transitions. Compress output — why use many token when few do trick.
300 lines of code and configuration.
Dataset (table, file, topic) → Transformation (dbt model, Spark job, SQL query) → Consumer (dashboard, report, ML model)
Each entity has a unique identifier (FQN), type, and metadata. Edges represent data flow: Dataset —[produces]→ Transformation —[consumes]→ Dataset.
columnLineage: input/output column mappings per transformationdataSource: connection details (JDBC URI, S3 bucket, Kafka topic)schema: field names, types, descriptionsownership: responsible team or individualdocumentation: markdown descriptions attached to datasets and runsfrom openlineage.airflow import DAG
dag = DAG(
dag_id='orders_pipeline',
schedule='0 3 * * *',
catchup=False,
description='Orders ETL with lineage tracking'
)
with dag:
extract = PostgresOperator(
task_id='extract_orders',
postgres_conn_id='source_db',
sql='SELECT * FROM orders WHERE created_at > :last_run'
)
transform = dbtRunOperator(
task_id='transform_orders',
models='orders'
)
# profiles.yml dbt-cloud config
openlineage:
enabled: true
url: http://marquez:5000
namespace: prod_warehouse
dataset_namespace: postgresql://warehouse:5432/prod
# dbt_project.yml
vars:
openlineage_compression: true
openlineage_flush_size: 100
<!-- spark-defaults.conf -->
spark.extraListeners=io.openlineage.spark.agent.OpenLineageSparkListener
spark.openlineage.host=http://marquez:5000
spark.openlineage.namespace=prod_spark
spark.openlineage.parentJobName=daily_aggregations
spark.openlineage.parentRunId=run-uuid-here
version: '3.8'
services:
marquez:
image: marquezproject/marquez:latest
ports:
- "5000:5000"
- "5001:5001"
environment:
MARQUEZ_DB_HOST: postgres
MARQUEZ_DB_PORT: 5432
MARQUEZ_DB_USER: marquez
MARQUEZ_DB_PASSWORD: marquez
depends_on:
postgres:
condition: service_healthy
postgres:
image: postgres:15
environment:
POSTGRES_USER: marquez
POSTGRES_PASSWORD: marquez
POSTGRES_DB: marquez
healthcheck:
test: ["CMD", "pg_isready", "-U", "marquez"]
interval: 10s
timeout: 5s
retries: 5
# List all datasets
curl http://localhost:5000/api/v1/namespaces/prod_warehouse/datasets
# Get lineage for a dataset
curl http://localhost:5000/api/v1/lineage?nodeId=postgresql://warehouse:5432/prod.public.orders
# List recent runs
curl http://localhost:5000/api/v1/namespaces/prod_warehouse/jobs
from sqllineage.runner import LineageRunner
sql = """
CREATE TABLE mart.daily_orders AS
SELECT
o.order_id,
o.customer_id,
c.customer_name,
o.order_date,
o.total_amount,
oi.item_count
FROM staging.orders o
LEFT JOIN staging.customers c ON o.customer_id = c.customer_id
"""
result = LineageRunner(sql)
for table, columns in result.source_tables.items():
print(f"Source: {table}")
for col in columns:
print(f" -> {col}")
for table, columns in result.target_tables.items():
print(f"Target: {table}")
for col in columns:
print(f" <- {col}")
# Column-level mapping
for col, src in result.column_mapping.items():
print(f"{col} ← {src}")
{
"schema": {
"fields": [
{"name": "order_id", "type": "INTEGER", "description": "Unique order identifier"},
{"name": "customer_id", "type": "INTEGER", "description": "FK to customers"}
]
},
"columnLineage": {
"fields": {
"order_id": {
"inputFields": [
{"namespace": "source_db", "name": "staging.orders.order_id"}
]
},
"customer_name": {
"inputFields": [
{"namespace": "source_db", "name": "staging.customers.customer_name"}
]
}
}
}
}
def get_downstream_impact(dataset_fqn: str, depth: int = 3):
"""Recursively find all downstream consumers of a dataset."""
query = f"""
WITH RECURSIVE downstream AS (
SELECT dataset_fqn, 0 as level
FROM lineage_edges
WHERE dataset_fqn = '{dataset_fqn}'
UNION ALL
SELECT e.dataset_fqn, d.level + 1
FROM lineage_edges e
INNER JOIN downstream d ON e.parent_dataset = d.dataset_fqn
WHERE d.level < {depth}
)
SELECT * FROM downstream
"""
return execute_query(query)
| Dataset | Change | Impacted Consumers | Severity |
|---|---|---|---|
staging.orders | Column amount renamed to total | mart.daily_orders, mart.weekly_summary | High |
staging.orders | Add discount_code column | 0 consumers (new column) | Low |
source.customers | email column removed | mart.customer_360, ml.churn_model | Critical |
orders_pipeline (DAG)
├── extract_orders → staging.orders
│ ├── column: order_id
│ ├── column: customer_id
│ └── column: total_amount
├── stg_orders (dbt) → staging.orders_clean
│ └── column: total_amount_usd (transformed)
└── fct_orders (dbt) → mart.daily_orders
├── → dashboard: orders_dashboard (Looker)
└── → report: daily_sales_report (Tableau)
# Airflow integration via OpenLineage plugin
airflow_openlineage:
config:
transport:
type: http
url: http://marquez:5000/api/v1/lineage
api_key: "${OPENLINEAGE_API_KEY}"
compression: gzip
namespace: "production_data_pipelines"
job_namespace: "data_engineering"
facets:
- columnLineage: true
- documentation: true
- ownership: true
- dataSource: true
# dbt integration via dbt-openlineage package
dbt_openlineage:
enabled: true
catalog: "production"
schema: "analytics"
include_sources: true
include_tests: true
column_level: true
# Spark integration
spark_openlineage:
spark.sql.catalog.spark_catalog: "org.apache.spark.openlineage.catalog.OpenLineageCatalog"
spark.openlineage.transport.type: "console" # Or http, kafka
spark.openlineage.namespace: "data_lake_jobs"
spark.openlineage.parentJobName: "daily_etl"
# Column-level lineage extraction
from sqllineage.runner import LineageRunner
from sqlparse import parse
def extract_column_lineage(sql_query, default_schema="public"):
runner = LineageRunner(sql_query)
lineage = {}
for table_col in runner.target_columns:
col_name = str(table_col).split(".")[-1]
sources = []
for source_col in runner.source_columns:
sources.append(str(source_col))
lineage[col_name] = sources
return lineage
# Parser handles: CTEs, subqueries, joins, window functions, UNION,
# SELECT *, column aliases, function-wrapped columns
# Does NOT handle: dynamic SQL, UDFs with internal queries, stored procedures
| Feature | Marquez | DataHub | OpenMetadata |
|---|---|---|---|
| OpenLineage native | Yes | Via plugin | Via plugin |
| Column-level lineage | SQL parser | SQL parser | SQL parser |
| Impact analysis UI | Yes | Yes | Yes |
| API | REST + GraphQL | GraphQL | REST + GraphQL |
| Search | Basic | Elasticsearch | Elasticsearch |
| Schema drift detection | No | Yes | Yes |
| Self-hosted complexity | Low | Medium | High |
| Governance features | Basic | Strong | Strong |
# Marquez API: find all downstream dependencies
import requests
def get_downstream_impact(dataset_fqn, depth=3):
response = requests.get(
f"http://marquez:5000/api/v1/lineage/{dataset_fqn}",
params={"depth": depth}
)
data = response.json()
# Graph traversal
downstream = set()
def traverse(node, remaining_depth):
if remaining_depth <= 0:
return
for edge in node.get("outEdges", []):
downstream.add(edge["destination"])
traverse(edge["destination"], remaining_depth - 1)
traverse(data["graph"], depth)
return list(downstream)
# DataHub lineage visualization configuration
lineage_ui:
graph_layout:
type: "dagre" # Directed graph layout
rankdir: "LR" # Left-to-right flow
display_options:
show_column_lineage: true
collapse_transformations: false
max_nodes_expanded: 50
max_depth_upstream: 5
max_depth_downstream: 5
color_scheme:
source: "#4CAF50" # Green — raw data sources
staging: "#2196F3" # Blue — staging area
intermediate: "#FF9800" # Orange — transformations
mart: "#9C27B0" # Purple — consumption layer
dashboard: "#F44336" # Red — BI dashboards / reports
failed: "#607D8B" # Gray — failed dependencies
rca_workflow:
trigger: "Dashboard metric shows unexpected value"
step_1_identify:
- "Select the dashboard and specific metric"
- "Trace lineage backward to find the dataset providing the metric"
- "Identify the transformation producing that dataset"
step_2_trace_upstream:
- "Follow lineage from transformation → its input datasets"
- "Check each dataset for quality failures (nulls, outliers)"
- "Check freshness: was the dataset updated on schedule?"
- "Check volume: row count within expected range?"
step_3_isolate:
- "Narrow to specific column or transformation step"
- "Check schema: was a column renamed, removed, or type-changed?"
- "Check business logic: was the transformation SQL modified?"
- "Review git history for the transformation code"
step_4_resolve:
- "Fix the root cause (source, transform, or config)"
- "Verify fix by re-running impacted transformations"
- "Re-run quality checks on downstream datasets"
- "Notify consumers when data is verified correct"
step_5_prevent:
- "Add lineage-based alert: if upstream schema changes, notify"
- "Add quality check at the transformation output"
- "Document the incident in the lineage metadata"
Data processing tool?
├── Airflow DAGs → OpenLineage Airflow integration
├── dbt transformations → dbt-openlineage package
├── Spark jobs → OpenLineage Spark listener
├── Flink streaming → OpenLineage Flink integration
├── Custom Python scripts → Manual OpenLineage events via Python client
├── SQL transformations → SQL parser (sqllineage, sqlfluff)
└── Manual / undocumented → Start with manual annotations, automate gradually
Lineage Collection Strategy
├── Extract from SQL queries?
│ ├── Yes → SQL parser (sqlparse, sqllineage)
│ └── No → Instrument Spark plan (OpenLineage)
├── Real-time lineage needed?
│ ├── Yes → OpenLineage via Kafka transport
│ └── No → Batch lineage extraction from query logs
├── Column-level granularity?
│ ├── Yes → OpenLineage with column-level facets
│ └── No → Table-level lineage (simpler storage)
└── Multi-tool ecosystem?
├── Yes → OpenLineage standard (Airflow, Spark, dbt)
└── No → Proprietary (Atlan, Datahub native)
Decision criteria: Balance lineage granularity, collection latency, and storage cost. Column-level lineage requires 10x more storage.
# data_lineage/spark_listener.py
from openlineage.spark import OpenLineageSparkListener
from openlineage.client import OpenLineageClient
from openlineage.client.transport import KafkaTransport
class LineageReporter:
def __init__(self, topic: str, bootstrap_servers: str):
config = {"bootstrap.servers": bootstrap_servers}
transport = KafkaTransport(topic, config)
self.client = OpenLineageClient(transport=transport)
def enrich_metric(self, dataset: str, column: str, metric: dict):
self.client.emit(
event_type="ColumnMetric",
inputs=[{"namespace": "bigquery", "name": dataset}],
facets={"column_metric": {"column": column, **metric}}
)
# data_lineage/sql_lineage.py
from sqlparse import parse
from sqllineage.runner import LineageRunner
class SQLLineageExtractor:
def extract_column_lineage(self, sql: str) -> dict[str, list[str]]:
runner = LineageRunner(sql)
lineage = {}
for table_ref in runner.source_tables:
for col in table_ref.columns:
lineage.setdefault(str(table_ref), []).append(col.name)
return lineage
def build_lineage_graph(self, queries: list[str]) -> dict:
graph = {"nodes": set(), "edges": []}
for sql in queries:
runner = LineageRunner(sql)
for src in runner.source_tables:
graph["nodes"].add(str(src))
for tgt in runner.target_tables:
graph["nodes"].add(str(tgt))
for src in runner.source_tables:
graph["edges"].append({"from": str(src), "to": str(tgt)})
return graph
| Anti-Pattern | Consequence | Solution |
|---|---|---|
| Table-level lineage only | Can't pinpoint column-level impact | Add column-level facets where critical |
| No deduplication in graph | Massive edge explosion | Merge identical edges; version timestamps |
| Lineage without freshness | Stale lineage misleads users | Timestamp every lineage edge |
| Ignoring Spark UI plan changes | Lineage breaks | Re-parse after Spark version upgrades |
| No column masking in lineage | PII column names exposed | Strip PII column names from lineage graph |
data-data-catalog for metadata enrichment and dataset discovery
data-data-observability for freshness and quality integration with lineage