Track data lineage across pipelines. Covers column-level lineage, OpenLineage, data catalogs, impact analysis, root cause analysis, and building lineage into your data stack.
Data lineage tracks the flow of data from source to destination, while observability provides visibility into the health and behavior of data pipelines. This becomes essential when debugging failures, understanding impact of schema changes, or validating compliance requirements. Lineage alone answers "where did this data come from?" while observability adds "how is it behaving now?"
Lineage tracking requires recording which inputs feed into a process and what outputs it produces. This can be expressed in SQL:
INSERT INTO lineage_graph (source_table, target_table, pipeline_id, processed_at)
VALUES ('sales_raw', 'sales_aggregated', 'etl-2023-11-01', NOW());Each pipeline execution must capture the mapping between input and output tables. The system stores these relationships in a graph structure where nodes represent data entities (tables, files) and edges represent transformations.
When data flows through multiple steps, each transformation must be recorded:
INSERT INTO lineage_edges (from_entity, to_entity, edge_type, pipeline_id)
VALUES
('sales_raw', 'sales_cleaned', 'transform', 'etl-2023-11-01'),
('sales_cleaned', 'sales_aggregated', 'aggregate', 'etl-2023-11-01');The edge_type field distinguishes between transformations (like filtering or joining) and aggregations. This enables queries to trace how data changes as it moves through the pipeline.
Some pipelines read from dynamic sources like partitioned files or time-based tables:
INSERT INTO lineage_partitions (table_name, partition_key, partition_value, pipeline_id)
VALUES ('sales_2023', 'date', '2023-11-01', 'etl-2023-11-01');These records allow lineage queries to understand which partitions of a table are used in a given pipeline run.
Observability requires capturing execution status, duration, and resource usage:
INSERT INTO pipeline_execution (
pipeline_id,
status,
start_time,
end_time,
duration_seconds,
rows_processed,
error_message
) VALUES (
'etl-2023-11-01',
'success',
'2023-11-01T01:00:00Z',
'2023-11-01T01:05:00Z',
300,
150000,
NULL
);The status field can contain values like success, failed, skipped, or warning. Error messages must be stored verbatim to enable debugging.
Data quality checks should be recorded alongside execution data:
INSERT INTO data_quality_metrics (
pipeline_id,
metric_name,
metric_value,
threshold_value,
status
) VALUES (
'etl-2023-11-01',
'null_count',
500,
100,
'warning'
);The metric_name field uses standard names like null_count, duplicate_count, or range_check. The status indicates whether the check passed, failed, or generated a warning.
To determine where data comes from, query the lineage graph:
SELECT source_table, pipeline_id
FROM lineage_graph
WHERE target_table = 'sales_aggregated'
ORDER BY processed_at DESC
LIMIT 1;This returns the most recent input table for a given output. For complex dependencies, follow the edge relationships recursively.
To understand how data flows through multiple steps:
WITH RECURSIVE lineage_tree AS (
SELECT from_entity, to_entity, edge_type, 0 as depth
FROM lineage_edges
WHERE to_entity = 'sales_aggregated'
UNION ALL
SELECT e.from_entity, e.to_entity, e.edge_type, lt.depth + 1
FROM lineage_edges e
JOIN lineage_tree lt ON e.to_entity = lt.from_entity
)
SELECT from_entity, edge_type, depth
FROM lineage_tree
ORDER BY depth;This recursive query traces back through the transformation chain, showing each step and its type.
To find all downstream tables affected by a change:
SELECT DISTINCT target_table
FROM lineage_graph
WHERE source_table = 'sales_raw';This shows all tables that depend on the raw sales data. For more complex dependencies involving intermediate transformations, use the edge table to trace forward.
Failed pipeline executions must be recorded with error details:
INSERT INTO pipeline_execution (
pipeline_id,
status,
start_time,
end_time,
duration_seconds,
error_message
) VALUES (
'etl-2023-11-01',
'failed',
'2023-11-01T01:00:00Z',
'2023-11-01T01:05:00Z',
300,
'Connection timeout to data warehouse'
);The error message must be captured exactly as returned by the system, including stack traces or specific error codes.
When pipelines fail, check if they also produced data quality issues:
SELECT
pe.pipeline_id,
pe.error_message,
dqm.metric_name,
dqm.metric_value
FROM pipeline_execution pe
JOIN data_quality_metrics dqm ON pe.pipeline_id = dqm.pipeline_id
WHERE pe.status = 'failed'
AND dqm.status = 'failed';This identifies when data quality issues coincide with execution failures.
When table schemas change, capture the modification:
INSERT INTO schema_changes (
table_name,
old_schema,
new_schema,
change_timestamp,
pipeline_id
) VALUES (
'sales_raw',
'{"id": "integer", "amount": "decimal"}',
'{"id": "integer", "amount": "decimal", "currency": "string"}',
'2023-11-01T01:00:00Z',
'etl-2023-11-01'
);The schema fields use JSON format to store complete definitions. Each change must be linked to the pipeline that caused it.
To see how schema changes affect downstream pipelines:
SELECT
lg.target_table,
sc.change_timestamp
FROM lineage_graph lg
JOIN schema_changes sc ON lg.source_table = sc.table_name
WHERE sc.table_name = 'sales_raw'
ORDER BY sc.change_timestamp;This shows which tables are affected by schema changes and when they occurred.
To track performance over time:
SELECT
pipeline_id,
AVG(duration_seconds) as avg_duration,
MAX(duration_seconds) as max_duration,
COUNT(*) as execution_count
FROM pipeline_execution
WHERE start_time >= '2023-11-01'
GROUP BY pipeline_id;This provides metrics for performance analysis and alerting.
To identify when quality thresholds are crossed:
SELECT
pipeline_id,
metric_name,
metric_value,
threshold_value
FROM data_quality_metrics
WHERE status = 'failed'
AND created_at >= NOW() - INTERVAL '1 hour';This enables real-time alerts when data quality issues arise.
The most common failure occurs when lineage recording is skipped during partial pipeline runs. If a pipeline only processes some partitions and fails to update the lineage graph, downstream systems may use stale data. This typically manifests as missing records in the lineage tables or incorrect dependencies in queries.
Pipeline failures often store error messages inconsistently. Some systems log errors in a separate table while others embed them directly in execution records. This inconsistency makes querying and alerting difficult.
When schema changes occur, pipeline executions may reference old and new schemas simultaneously. If lineage tracking only captures one version, queries can return incorrect results about which data was used in a given run.
Timestamps in lineage records must be stored consistently. Inconsistent timezone handling causes queries to miss data or show incorrect relationships between entities.
The following minimal schema is sufficient for basic lineage:
CREATE TABLE lineage_graph (
source_table VARCHAR(255),
target_table VARCHAR(255),
pipeline_id VARCHAR(255),
processed_at TIMESTAMP
);
CREATE TABLE pipeline_execution (
pipeline_id VARCHAR(255) PRIMARY KEY,
status VARCHAR(50),
start_time TIMESTAMP,
end_time TIMESTAMP,
duration_seconds INTEGER,
rows_processed INTEGER,
error_message TEXT
);Database users need INSERT permissions on lineage and execution tables. Read permissions are required for querying lineage information.
Old lineage records should be archived or purged based on retention policies. This prevents the lineage database from growing indefinitely while maintaining historical relationships.
The key to effective data lineage and observability is consistency in recording information. Each pipeline execution must capture the same set of relationships, metrics, and status information. When this consistency breaks, debugging becomes significantly more difficult and reliability decreases.
This page was rewritten on 15 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.
We use cookies for analytics (Google Analytics) and advertising (Google AdSense) to improve your experience and support free content. Privacy Policy