Data Lineage — Tracking Data Flow, Impact Analysis and Governance
In this tutorial, you'll learn about Data Lineage. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Data lineage tracks the complete lifecycle of data from its origin through transformations, aggregations, and storage to its final consumption in reports, dashboards, and Machine Learning models.
What You'll Learn
You'll master lineage capture at table, column, and row levels, impact analysis for schema changes and pipeline modifications, OpenLineage standard integration with Apache Spark and Airflow, column-level lineage with SQL parsers, and lineage-driven data governance for regulatory Compliance.
Why It Matters
Data lineage answers critical questions: "Where did this number come from?" "What reports will break if I change this schema?" "Is this data compliant with GDPR retention policies?" Without lineage, data teams spend 30 percent of their time on manual data discovery and debugging. At DodaTech, lineage tracking in Durga Antivirus Pro's threat analysis pipeline ensures every security alert can be traced to its raw log source.
Real-World Use
A finance analyst sees a 5 percent drop in Q4 revenue in the executive dashboard. With data lineage, they trace the number back through the dashboard SQL view, through the dbt transformation model, through the Airflow pipeline, to the raw source table, and discover that a new data source filter was incorrectly excluding international transactions. Without lineage, this debugging takes days instead of minutes.
Data Lineage Architecture
flowchart LR
Source[(Source DB)] --> Extractor[Extractor]
Extractor --> Lineage[Lineage Event Producer]
Lineage --> Kafka[Kafka]
Kafka --> Collector[Lineage Collector]
Collector --> Backend[(Lineage Store
Neo4j + PostgreSQL)]
Backend --> API[Lineage API]
API --> UI[Lineage UI / Graph]
API --> Impact[Impact Analyzer]
API --> Catalog[Data Catalog Integration]
Transform[Spark/Dbt/Airflow] -->|OpenLineage Events| Kafka
OpenLineage Standard
OpenLineage is an open standard for lineage metadata collection. It defines a common schema for run events, dataset inputs/outputs, and job definitions.
{
"eventType": "COMPLETE",
"eventTime": "2026-06-22T10:30:00Z",
"run": {
"runId": "d4812b3e-1f4c-4a7f-8c2d-9e3f0a1b2c3d",
"facets": {
"spark.logicalPlan": {
"_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.0/integration/spark",
"_schemaURL": "https://openlineage.io/spec/1-0-0/SparkLogicalPlanFacet",
"plan": "[Project [revenue, region, quarter]]"
}
}
},
"job": {
"namespace": "dodatech-etl",
"name": "finance.revenue_aggregation.daily",
"facets": {
"documentation": {
"_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.0/integration/airflow",
"description": "Aggregates daily revenue by region and quarter"
}
}
},
"inputs": [{
"namespace": "postgres://analytics-db:5432",
"name": "public.transactions",
"facets": {
"schema": {
"fields": [
{"name": "amount", "type": "decimal"},
{"name": "region", "type": "varchar"},
{"name": "transaction_date", "type": "timestamp"}
]
}
}
}],
"outputs": [{
"namespace": "postgres://analytics-db:5432",
"name": "public.daily_revenue",
"facets": {
"schema": {
"fields": [
{"name": "revenue", "type": "decimal"},
{"name": "region", "type": "varchar"},
{"name": "quarter", "type": "varchar"}
]
}
}
}]
}
Column-Level Lineage with SQL Parsing
import sqlparse
from sqlparse.sql import Identifier, Comparison
from collections import defaultdict
class ColumnLineageParser:
def __init__(self):
self.lineage_map = defaultdict(list)
def parse_select(self, sql: str, source_table: str):
parsed = sqlparse.parse(sql)[0]
if parsed.get_type() != 'SELECT':
return
# Extract SELECT columns and their sources
select_clause = None
from_clause = None
for token in parsed.tokens:
if token.ttype is None and isinstance(token, sqlparse.sql.IdentifierList):
select_clause = token
if token.ttype is None and token.get_type() == 'FROM':
from_clause = token
if not select_clause or not from_clause:
return
# Map each output column to its source
for column in select_clause.get_identifiers():
col_name = column.get_name()
if column.has_alias():
target_col = column.get_alias()
else:
target_col = col_name
# Trace through expressions
if isinstance(column, sqlparse.sql.Function):
# Extract source columns from function arguments
sources = self._extract_column_refs(column)
self.lineage_map[(source_table, target_col)] = sources
else:
self.lineage_map[(source_table, target_col)] = [col_name]
def _extract_column_refs(self, token):
refs = []
if isinstance(token, Identifier):
refs.append(token.get_name())
elif hasattr(token, 'tokens'):
for t in token.tokens:
refs.extend(self._extract_column_refs(t))
return refs
def build_lineage_graph(self):
graph = defaultdict(set)
for (table, col), sources in self.lineage_map.items():
for src in sources:
graph[(table, col)].add(src)
return graph
lineage = ColumnLineageParser()
lineage.parse_select(
"""SELECT amount, region,
CASE WHEN amount > 1000 THEN 'high' ELSE 'standard' END AS tier
FROM raw_transactions""",
"daily_revenue"
)
print(dict(lineage.lineage_map))
Expected behavior: The parser identifies that daily_revenue.amount comes from raw_transactions.amount, daily_revenue.region comes from raw_transactions.region, and daily_revenue.tier is derived from raw_transactions.amount with a CASE expression. This enables impact analysis: changing the amount column in raw_transactions affects both amount and tier in daily_revenue.
Impact Analysis for Schema Changes
def analyze_schema_change(column_path: str, lineage_graph: dict) -> list:
impacted = []
def traverse(current, path):
for (table, col), sources in lineage_graph.items():
if col == current and table not in path:
impacted.append(f"{table}.{col}")
traverse(col, path + [table])
traverse(column_path.split(".")[1], [])
return impacted
lineage_graph = {
("raw_transactions", "amount"): set(),
("daily_revenue", "revenue"): {"amount"},
("quarterly_report", "total_revenue"): {"revenue"},
("exec_dashboard", "q4_revenue"): {"total_revenue"},
}
impacted = analyze_schema_change("raw_transactions.amount", lineage_graph)
print(f"Impacted by changing amount: {impacted}")
Expected behavior: Changing raw_transactions.amount propagates through the lineage chain: daily_revenue.revenue is impacted, quarterly_report.total_revenue is impacted, and exec_dashboard.q4_revenue is impacted. The team knows exactly which dashboards and reports need updates.
Lineage Store in Neo4j
Neo4j is ideal for lineage storage because lineage is inherently a graph.
// Create lineage nodes
CREATE (trans:Dataset {name: 'raw_transactions'})
CREATE (revenue:Dataset {name: 'daily_revenue'})
CREATE (job:Job {name: 'revenue_aggregation', type: 'dbt'})
// Link datasets through jobs
CREATE (trans)-[:PRODUCES]->(job)
CREATE (job)-[:CONSUMES]->(revenue)
// Add column-level lineage
MATCH (trans:Dataset {name: 'raw_transactions'})
MATCH (revenue:Dataset {name: 'daily_revenue'})
CREATE (trans)-[:COLUMN_MAP {
source_col: 'amount',
target_col: 'revenue',
transform: 'SUM(amount)'
}]->(revenue)
// Query: find all upstream dependencies for a column
MATCH path = (target:Dataset {name: 'exec_dashboard'})
<-[:CONSUMES*]-
(upstream)
RETURN path
Common Errors
1. Table-Level Only Lineage
Tracking only table-level lineage misses critical column-level dependencies. A schema change to a single column requires column-level lineage to determine true impact.
2. No Automated Capture
Manual lineage documentation is always outdated. Automate capture using OpenLineage integrations with Spark, dbt, Airflow, and SQL databases.
3. Ignoring Transformation Logic
Lineage without transformation information shows where data comes from but not how it was changed. Include transform expressions, filtering conditions, and aggregation logic in lineage events.
4. No Versioning
Lineage changes over time as pipelines evolve. Store lineage per pipeline run version so historical data can be traced to the correct schema state.
5. Graph Database for Everything
Not every lineage query needs a graph DB. Store recent lineage in PostgreSQL for fast API queries and use Neo4j for deep recursive traversal.
6. Missing Downstream Notifications
When a source schema changes, affected teams should be notified automatically. Implement a subscription system where dataset owners get alerts when their upstream dependencies change.
7. Over-Capturing
Capturing lineage at the row level for every record creates massive storage costs. Capture column-level lineage by default and row-level lineage for regulated data only.
Practice Questions
1. What is the difference between table-level and column-level lineage?
Table-level lineage shows which tables feed into which other tables. Column-level lineage tracks individual columns through transformations, showing that revenue in the dashboard comes from the amount column in raw_transactions summed and grouped by region.
2. Why is lineage important for GDPR Compliance?
GDPR's right to erasure requires deleting all personal data across the entire data pipeline. Lineage shows where PII data flows, enabling complete deletion rather than partial removal from a single system.
3. How does OpenLineage standardize lineage collection?
OpenLineage defines a common JSON schema for run events, dataset facets, and job facets. Any tool implementing the standard (Spark, Airflow, dbt, Flink) produces lineage events in the same format, enabling a single lineage collector.
4. Why use a graph database for lineage storage?
Lineage is inherently a directed acyclic graph. Graph databases optimize recursive traversal queries like "find all upstream dependencies" that would require multiple recursive self-joins in SQL.
5. Challenge: Design a lineage system for a real-time streaming pipeline. Track column-level lineage from Kafka topics through Flink transformations to Elasticsearch indexes and Redis caches. Handle the challenge that streaming transformations produce continuous output without discrete runs.
Mini Project: Lineage Impact Analyzer
Build a lineage impact analysis tool:
- Accept OpenLineage events via HTTP POST (simulate with sample events)
- Store lineage in a Neo4j instance (or in-memory graph for simplicity)
- Provide an API: GET /impact?column=raw_transactions.amount
- Return all downstream columns and datasets with their distance from the source
- Provide a web UI showing the lineage graph with color coding for impact depth
- Add a subscription endpoint where teams can subscribe to change notifications for specific datasets
Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro