Production engineering guide for data lineage tracking covering patterns, implementation strategies, and operational best practices.
Data lineage tracking captures the flow of data from source to destination, documenting how datasets are transformed, joined, and consumed across systems. This becomes essential when debugging data issues, understanding impact of schema changes, or auditing compliance requirements. The process involves recording technical metadata about data movement and transformations, enabling engineers to trace back from a final dataset to its origins and forward to all downstream consumers.
To track lineage for ETL processes, you must record each transformation step with source, target, and operation details. Most systems use a metadata table or database to store this information.
CREATE TABLE data_lineage (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
source_dataset TEXT NOT NULL,
target_dataset TEXT NOT NULL,
operation_type TEXT NOT NULL,
operation_details JSONB,
execution_time TIMESTAMP DEFAULT NOW(),
job_name TEXT,
version TEXT
);The operation_details field typically captures parameters that define the transformation. For example, when performing a SQL join:
INSERT INTO data_lineage (
source_dataset,
target_dataset,
operation_type,
operation_details,
job_name
) VALUES (
'sales_raw',
'sales_enriched',
'JOIN',
'{"join_columns": ["customer_id"], "join_type": "LEFT", "source_tables": ["sales_raw", "customers"]}',
'daily_sales_job'
);Tracking data movement between systems requires explicit logging of transfers, including source location, destination, and metadata about the transfer. For batch jobs, this usually means recording job execution details.
For systems using a queue-based approach, each message should carry lineage information:
{
"lineage": {
"source_job": "customer_import",
"source_dataset": "customer_raw",
"version": "2023-10-01",
"timestamp": "2023-10-01T10:00:00Z"
},
"data": {
"customer_id": "CUST-001",
"name": "John Doe"
}
}To find all datasets that depend on a specific source, you can query the lineage table:
SELECT DISTINCT target_dataset FROM data_lineage
WHERE source_dataset = 'customer_raw';To trace back from a final dataset to its origins:
WITH RECURSIVE lineage_tree AS (
SELECT source_dataset, target_dataset, operation_type, 1 as level
FROM data_lineage
WHERE target_dataset = 'sales_final'
UNION ALL
SELECT d.source_dataset, d.target_dataset, d.operation_type, lt.level + 1
FROM data_lineage d
JOIN lineage_tree lt ON d.target_dataset = lt.source_dataset
)
SELECT * FROM lineage_tree ORDER BY level DESC;When transformations involve multiple steps or are composed of sub-transformations, you must track the relationship between these components. For instance, a single ETL job might include several SQL operations:
-- Job: customer_enrichment_pipeline
INSERT INTO data_lineage (
source_dataset,
target_dataset,
operation_type,
operation_details,
job_name
) VALUES
('customer_raw', 'customer_cleaned', 'TRANSFORM', '{"script": "remove_nulls()", "columns": ["email"]}', 'customer_enrichment_pipeline'),
('customer_cleaned', 'customer_enriched', 'JOIN', '{"join_columns": ["account_id"], "source_tables": ["customer_cleaned", "accounts"]}', 'customer_enrichment_pipeline');Lineage tracking should include quality metrics and validation information. This allows you to understand when and how data quality issues propagate:
ALTER TABLE data_lineage ADD COLUMN quality_metrics JSONB;You might record data quality checks that occurred during transformation:
INSERT INTO data_lineage (
source_dataset,
target_dataset,
operation_type,
operation_details,
quality_metrics
) VALUES (
'sales_raw',
'sales_cleaned',
'TRANSFORM',
'{"script": "validate_dates()", "columns": ["order_date"]}',
'{"row_count": 10000, "null_values": {"order_date": 50}, "validation_errors": 3}'
);When datasets change schema, lineage tracking must capture both the old and new versions. You can do this by maintaining versioned references:
ALTER TABLE data_lineage ADD COLUMN source_version TEXT;
ALTER TABLE data_lineage ADD COLUMN target_version TEXT;For example, if you have a dataset that was updated from version 1 to version 2:
INSERT INTO data_lineage (
source_dataset,
target_dataset,
operation_type,
operation_details,
source_version,
target_version
) VALUES (
'user_events_v1',
'user_events_v2',
'SCHEMA_CHANGE',
'{"changes": ["added column timestamp", "renamed column user_id to customer_id"]}',
'1.0',
'2.0'
);In production, you should log lineage information with every transformation that modifies data. This requires integration into your existing batch and streaming pipelines.
For a Python-based ETL job:
import json
import psycopg2
def log_lineage(source_dataset, target_dataset, operation_type, operation_details, job_name):
conn = psycopg2.connect(
host="postgres.example.com",
database="lineage_db",
user="lineage_user",
password="secret"
)
cur = conn.cursor()
cur.execute("""
INSERT INTO data_lineage (source_dataset, target_dataset, operation_type, operation_details, job_name)
VALUES (%s, %s, %s, %s, %s)
""", (source_dataset, target_dataset, operation_type, json.dumps(operation_details), job_name))
conn.commit()
cur.close()
conn.close()
# Usage
log_lineage(
"sales_raw",
"sales_enriched",
"JOIN",
{"join_columns": ["customer_id"], "join_type": "LEFT"},
"daily_sales_job"
)A common failure mode is incomplete lineage logging. If a transformation fails partway through, you might have lineage entries for some steps but not others. This creates gaps in your traceability:
-- This query will show jobs with incomplete lineage
SELECT job_name, COUNT(*) as step_count
FROM data_lineage
GROUP BY job_name
HAVING COUNT(*) < (
SELECT COUNT(*)
FROM data_lineage dl2
WHERE dl2.job_name = data_lineage.job_name
AND dl2.operation_type IN ('JOIN', 'TRANSFORM')
);Another failure mode occurs when you have circular dependencies in lineage. A dataset might appear to depend on itself:
-- Detect self-referencing lineage entries
SELECT *
FROM data_lineage
WHERE source_dataset = target_dataset;You can integrate lineage tracking with version control systems like Git. This allows you to track when and why data flows changed, linking them to code commits:
# Example of adding lineage information to a commit message
git commit -m "Update customer pipeline schema
Lineage: customer_raw -> customer_enriched (JOIN)
Job: customer_pipeline_v2
Version: 1.2"To understand what datasets will be affected by a schema change, you can query the lineage tree:
WITH RECURSIVE downstream AS (
SELECT target_dataset, source_dataset, operation_type, 1 as level
FROM data_lineage
WHERE source_dataset = 'customer_raw'
UNION ALL
SELECT d.target_dataset, d.source_dataset, d.operation_type, down.level + 1
FROM data_lineage d
JOIN downstream down ON d.source_dataset = down.target_dataset
)
SELECT DISTINCT target_dataset FROM downstream;Many systems provide tools to visualize lineage relationships. You can represent this as a graph where nodes are datasets and edges represent transformations:
# Example of exporting lineage for visualization
psql -d lineage_db -c "
SELECT
source_dataset,
target_dataset,
operation_type,
execution_time
FROM data_lineage
ORDER BY execution_time
" > lineage_export.csvTo validate that your lineage tracking is complete, you can check for missing relationships:
-- Find datasets that have no upstream dependencies
SELECT DISTINCT source_dataset
FROM data_lineage
WHERE source_dataset NOT IN (
SELECT target_dataset FROM data_lineage
)
AND source_dataset != 'source';Lineage information should be maintained over time. You can create retention policies:
-- Remove lineage entries older than 90 days
DELETE FROM data_lineage
WHERE execution_time < NOW() - INTERVAL '90 days';One frequent mistake is not logging all transformations. If a system processes data through multiple steps but only logs the final output, you lose visibility into intermediate transformations. This happens when the lineage tracking code is placed only at the end of pipelines rather than in each transformation step.
Another issue occurs with parallel processing where multiple jobs might write to the same target dataset. In such cases, you must ensure that lineage entries correctly reflect all contributing jobs:
-- Example: Multiple jobs updating the same dataset
INSERT INTO data_lineage (source_dataset, target_dataset, operation_type, job_name)
VALUES
('sales_jan', 'sales_monthly', 'MERGE', 'jan_sales_job'),
('sales_feb', 'sales_monthly', 'MERGE', 'feb_sales_job');The sharp edge here is that if you don't track which jobs contributed to a merge, it's difficult to understand why a dataset changed.
In production environments, lineage tracking should not significantly impact performance. You might batch insert lineage entries:
def batch_log_lineage(entries):
conn = psycopg2.connect(
host="postgres.example.com",
database="lineage_db",
user="lineage_user",
password="secret"
)
cur = conn.cursor()
execute_values(cur, """
INSERT INTO data_lineage (source_dataset, target_dataset, operation_type, operation_details, job_name)
VALUES %s
""", entries)
conn.commit()
cur.close()
conn.close()This approach reduces the number of database round trips required for lineage logging.
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