How to Track Data Pipeline Dependencies Automatically with DataLineage

---
title: "Your Pipeline is Lying to You (And Here's How to Catch It)"
published: false
tags: [dataengineering, python, tutorial, dataquality]
---

# Your Pipeline is Lying to You (And Here's How to Catch It)

Last Tuesday at 2 AM, a Slack message woke up a data engineer somewhere. A dashboard was broken. The on-call scrambled through dbt models, Airflow DAGs, and three Spark jobs trying to answer one question: *who touched what?*

Four hours later, they found it. A column rename. Innocent. Undocumented.

This tutorial is about never having that 2 AM call again.

We'll use **DataLineage** — a tool that automatically maps dependencies across your entire pipeline stack — to build a system that *knows* what breaks before you deploy the thing that breaks it.

---

## What We're Actually Building

By the end of this post, you'll have a Python workflow that:

1. Traces a schema change through your entire pipeline
2. Identifies every downstream consumer affected
3. Runs an impact assessment *before* you merge that PR

No more archaeological digs through YAML files at midnight.

---

## Prerequisites

bash
pip install requests python-dotenv

You'll need a DataLineage API key. Set it in a `.env` file:

properties
DATALINEAGE_API_KEY=your_key_here
DATALINEAGE_BASE_URL=https://api.datalineage.io/v1

Let's write a small client wrapper we'll reuse throughout:

python
import os
import requests
from dotenv import load_dotenv

load_dotenv()

class DataLineageClient:
def init(self):
self.base_url = os.getenv("DATALINEAGE_BASE_URL")
self.headers = {
"Authorization": f"Bearer {os.getenv('DATALINEAGE_API_KEY')}",
"Content-Type": "application/json"
}

def _handle_response(self, response: requests.Response) -> dict:
    """Centralized error handling — because silent failures are the worst kind."""
    try:
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        error_body = response.json().get("error", "No error detail provided")
        raise RuntimeError(
            f"DataLineage API error [{response.status_code}]: {error_body}"
        ) from e
    return response.json()

> **Best practice:** Always surface the API's error message, not just the status code. "422" tells you nothing. "Column 'user_id' not found in registered schema" tells you everything.

---

## Step 1: Trace a Node Through the Pipeline

Every table, model, or dataset in DataLineage is a **node**. When you want to understand a node's full family tree — what it depends on, what depends on it — you call `POST /lineage/trace`.

Let's trace our `orders` dbt model:

python
def trace_lineage(client: DataLineageClient, node_id: str, depth: int = 3) -> dict:
"""
Trace upstream and downstream dependencies for a given node.

Args:
    node_id: The unique identifier for your table/model/dataset
    depth:   How many hops to traverse (default 3 is usually enough)
"""
payload = {
    "node_id": node_id,
    "depth": depth,
    "include_metadata": True  # Grab schema info, owner, last modified
}

response = client._handle_response(
    requests.post(
        f"{client.base_url}/lineage/trace",
        json=payload,
        headers=client.headers
    )
)
return response

Run it

client = DataLineageClient()
lineage = trace_lineage(client, node_id="dbt.production.orders")

print(f"Node: {lineage['node']['name']}")
print(f"Upstream dependencies: {len(lineage['upstream'])}")
print(f"Downstream consumers: {len(lineage['downstream'])}")

**Expected output:**

plaintext
Node: dbt.production.orders
Upstream dependencies: 4
Downstream consumers: 11

Eleven downstream consumers. That's eleven things that could break if someone renames a column. Let's see exactly what they are.

---

## Step 2: Retrieve and Parse the Full Lineage Graph

The trace job runs asynchronously for large graphs. Use the returned `lineage_id` to fetch results:

python
import time

def get_lineage_result(client: DataLineageClient, lineage_id: str, timeout: int = 30) -> dict:
"""
Poll for lineage results with a timeout.
Large graphs can take a few seconds to compute.
"""
deadline = time.time() + timeout

while time.time() < deadline:
    response = client._handle_response(
        requests.get(
            f"{client.base_url}/lineage/{lineage_id}",
            headers=client.headers
        )
    )

    status = response.get("status")

    if status == "complete":
        return response["graph"]
    elif status == "failed":
        raise RuntimeError(f"Lineage computation failed: {response.get('reason')}")

    # Still processing — wait and retry
    time.sleep(2)

raise TimeoutError(f"Lineage graph not ready after {timeout}s. Try increasing timeout.")

def print_dependency_tree(graph: dict) -> None:
"""Human-readable summary of what we found."""
print("\n📊 LINEAGE SUMMARY")
print("=" * 50)

print("\n⬆️  UPSTREAM (this node depends on):")
for node in graph.get("upstream", []):
    tool = node["metadata"].get("tool", "unknown")
    print(f"  └─ [{tool.upper()}] {node['name']}")

print("\n⬇️  DOWNSTREAM (depends on this node):")
for node in graph.get("downstream", []):
    tool = node["metadata"].get("tool", "unknown")
    owner = node["metadata"].get("owner", "unassigned")
    print(f"  └─ [{tool.upper()}] {node['name']}  (owner: {owner})")

Fetch and display

lineage_id = lineage["lineage_id"]
graph = get_lineage_result(client, lineage_id)
print_dependency_tree(graph)

**Expected output:**

plaintext

📊 LINEAGE SUMMARY

⬆️ UPSTREAM (this node depends on):
└─ [DBT] dbt.production.raw_orders
└─ [DBT] dbt.production.customers
└─ [SPARK] spark.etl.order_enrichment
└─ [AIRFLOW] airflow.dag.payment_sync

⬇️ DOWNSTREAM (depends on this node):
└─ [DBT] dbt.production.revenue_monthly (owner: [e-mail])
└─ [SPARK] spark.ml.churn_features (owner: [e-mail])
└─ [DBT] dbt.production.executive_kpis (owner: [e-mail])
... and 8 more

Now you can see the blast radius. The ML team's churn model feeds off `orders`. So does the executive KPI dashboard. One column rename, two very unhappy teams.

---

## Step 3: Run an Impact Assessment Before You Merge

This is the step that turns DataLineage from a debugging tool into a **prevention** tool.

Before any schema change, call `POST /lineage/impact` with the proposed change:

python
def assess_schema_change_impact(
client: DataLineageClient,
node_id: str,
proposed_changes: list[dict]
) -> dict:
"""
Simulate a schema change and identify every affected downstream consumer.

proposed_changes format:
    [{"type": "rename_column", "from": "old_name", "to": "new_name"}]
    [{"type": "drop_column",   "name": "column_name"}]
    [{"type": "change_type",   "column": "col", "from": "INT", "to": "STRING"}]
"""
payload = {
    "node_id": node_id,
    "changes": proposed_changes,
    "severity_threshold": "low"  # Catch even minor breakages
}

response = client._handle_response(
    requests.post(
        f"{client.base_url}/lineage/impact",
        json=payload,
        headers=client.headers
    )
)
return response

def format_impact_report(impact: dict) -> None:
"""Print a CI-friendly impact report."""
affected = impact.get("affected_nodes", [])
severity_counts = {"high": 0, "medium": 0, "low": 0}

print("\n🔍 IMPACT ASSESSMENT REPORT")
print("=" * 50)
print(f"Proposed change: {impact['change_summary']}")
print(f"Total affected nodes: {len(affected)}\n")

for node in affected:
    severity = node["impact_severity"]
    severity_counts[severity] += 1
    icon = {"high": "🔴", "medium": "🟡", "low": "🟢"}.get(severity, "⚪")

    print(f"{icon} [{severity.upper()}] {node['name']}")
    print(f"   Reason: {node['break_reason']}")
    print(f"   Owner:  {node['metadata'].get('owner', 'unknown')}\n")

print("SEVERITY SUMMARY:")
for level, count in severity_counts.items():
    print(f"  {level.capitalize()}: {count}")

# Exit non-zero if high-severity issues exist — useful in CI pipelines
if severity_counts["high"] > 0:
    print("\n❌ High-severity impacts detected. Review before merging.")
    return False

print("\n✅ No high-severity impacts. Proceed with caution.")
return True

Simulate renaming 'customer_id' to 'user_id' in orders

impact = assess_schema_change_impact(
client,
node_id="dbt.production.orders",
proposed_changes=[
{"type": "rename_column", "from": "customer_id", "to": "user_id"}
]
)

safe_to_merge = format_impact_report(impact)

**Expected output:**

plaintext

🔍 IMPACT ASSESSMENT REPORT

Proposed change: Rename column 'customer_id' → 'user_id' in dbt.production.orders
Total affected nodes: 7

🔴 [HIGH] spark.ml.churn_features
Reason: Column 'customer_id' referenced in JOIN condition (line 34)
Owner: [e-mail]

🔴 [HIGH] dbt.production.executive_kpis
Reason: Column 'customer_id' used in GROUP BY clause
Owner: [e-mail]

🟡 [MEDIUM] airflow.dag.weekly_report
Reason: Column referenced in downstream SELECT *
Owner: [e-mail]

SEVERITY SUMMARY:
High: 2
Medium: 1
Low: 4

❌ High-severity impacts detected. Review before merging.

---

## Putting It All Together: A Pre-Merge Check Script

Here's a complete script you can drop into your CI pipeline:

python

!/usr/bin/env python3

"""
pre_merge_lineage_check.py

Usage: python pre_merge_lineage_check.py --node dbt.production.orders \
--change rename_column --from customer_id --to user_id
"""

import argparse
import sys

def main():
parser = argparse.ArgumentParser(description="DataLineage pre-merge impact check")
parser.add_argument("--node", required=True)
parser.add_argument("--change", required=True, choices=["rename_column", "drop_column", "change_type"])
parser.add_argument("--from", dest="from_name", required=True)
parser.add_argument("--to", dest="to_name")
args = parser.parse_args()

client = DataLineageClient()

change = {"type": args.change, "from": args.from_name}
if args.to_name:
    change["to"] = args.to_name

print(f"🔎 Assessing impact of {args.change} on {args.node}...")

impact = assess_schema_change_impact(client, args.node, [change])
safe = format_impact_report(impact)

sys.exit(0 if safe else 1)  # Non-zero exit blocks the merge in CI

if name == "main":
main()

Add this to your GitHub Actions workflow and schema changes *cannot* merge without an impact review.

---

## The Bigger Picture

Manual dependency tracing is archaeology. You dig through layers of YAML, Spark configs, and Airflow DAGs hoping you've found everything — knowing you probably haven't.

DataLineage makes the invisible visible. Your pipeline has a graph structure whether you document it or not. This just makes it queryable.

The 2 AM Slack message? It becomes a PR comment: *"This rename affects 7 nodes. Here are the owners. Here's why it breaks."*

Ship with confidence. Your on-call engineer will thank you.

---

*Have a horror story about undocumented pipeline dependencies? Drop it in the comments — misery loves company.*