How to Track Data Pipeline Dependencies Automatically with DataLineage

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

# Your Data Pipeline is Lying to You (And How to Catch It)

Picture this: it's 9:47 AM on a Tuesday. A product manager pings you. The revenue dashboard is showing negative numbers. Your Slack is on fire. You *know* someone changed something upstream — but your pipeline spans dbt models, three Airflow DAGs, and a Spark job written by someone who left the company in 2022.

You are now a data archaeologist. You have a flashlight and no map.

This is the problem DataLineage solves. Instead of manually grep-ing through YAML files and interrogating git blame, you get an automated dependency graph across your *entire* stack. Let's build something real with it.

---

## What We're Actually Building

We'll simulate a realistic scenario: a `users` table schema change propagates through a pipeline, and we use DataLineage to find every downstream consumer *before* anyone's dashboard breaks.

Here's our fake-but-believable stack:
- A dbt model that reads from `users`
- An Airflow DAG that reads from that dbt model
- A custom ETL script that nobody documented

---

## Step 0: Setup

bash
pip install datalineage-sdk requests python-dotenv

python

config.py

import os
from dotenv import load_dotenv

load_dotenv()

BASE_URL = "https://api.datalineage.io/v1"
API_KEY = os.getenv("DATALINEAGE_API_KEY")

HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}

> **Best practice:** Never hardcode your API key. Future-you will thank present-you.

---

## Step 1: Register Your Pipeline Assets

Before we can trace anything, DataLineage needs to know what exists. Think of this as drawing the map before you need to use it.

python

register_assets.py

import requests
import json
from config import BASE_URL, HEADERS

def register_lineage(source: dict, destination: dict, pipeline_type: str, metadata: dict = None) -> dict:
"""
Register a dependency relationship between two pipeline assets.

Args:
    source: The upstream asset (table, model, topic)
    destination: The downstream consumer
    pipeline_type: One of 'dbt', 'airflow', 'spark', 'custom'
    metadata: Optional dict with owner, schedule, criticality, etc.

Returns:
    Lineage registration response with tracking ID
"""
payload = {
    "source": source,
    "destination": destination,
    "pipeline_type": pipeline_type,
    "metadata": metadata or {}
}

response = requests.post(
    f"{BASE_URL}/lineage/trace",
    headers=HEADERS,
    json=payload
)

# Don't silently swallow errors — surface them with context
if response.status_code != 201:
    raise RuntimeError(
        f"Failed to register lineage [{response.status_code}]: "
        f"{response.json().get('message', 'Unknown error')}\n"
        f"Payload was: {json.dumps(payload, indent=2)}"
    )

return response.json()

Register our three pipeline relationships

assets = [
{
"source": {"type": "table", "name": "raw.users", "database": "production"},
"destination": {"type": "dbt_model", "name": "marts.user_metrics", "project": "analytics"},
"pipeline_type": "dbt",
"metadata": {"owner": "data-team", "criticality": "high"}
},
{
"source": {"type": "dbt_model", "name": "marts.user_metrics", "project": "analytics"},
"destination": {"type": "airflow_dag", "name": "revenue_report_dag", "task": "aggregate_metrics"},
"pipeline_type": "airflow",
"metadata": {"owner": "analytics-eng", "schedule": "0 6 * * *"}
},
{
"source": {"type": "table", "name": "raw.users", "database": "production"},
"destination": {"type": "custom_etl", "name": "legacy_export_script", "repo": "data-infra"},
"pipeline_type": "custom",
"metadata": {"owner": "unknown", "criticality": "medium", "note": "The Greg Script™"}
}
]

lineage_ids = []
for asset in assets:
result = register_lineage(**asset)
lineage_ids.append(result["lineage_id"])
print(f"✓ Registered: {asset['source']['name']} → {asset['destination']['name']}")
print(f" Lineage ID: {result['lineage_id']}\n")

**Expected output:**

plaintext
✓ Registered: raw.users → marts.user_metrics
Lineage ID: lin_7f3a9c2e

✓ Registered: marts.user_metrics → revenue_report_dag
Lineage ID: lin_8b1d4f7a

✓ Registered: raw.users → legacy_export_script
Lineage ID: lin_2c9e5a3d

---

## Step 2: Retrieve and Inspect a Lineage Graph

Now let's see what DataLineage actually knows about `raw.users`. This is the "draw me the map" call.

python

inspect_lineage.py

import requests
from config import BASE_URL, HEADERS

def get_lineage(lineage_id: str, depth: int = 3) -> dict:
"""
Retrieve the full dependency graph for a registered lineage node.

Args:
    lineage_id: The ID returned from /lineage/trace
    depth: How many hops downstream to traverse (default: 3)

Returns:
    Full lineage graph with nodes and edges
"""
response = requests.get(
    f"{BASE_URL}/lineage/{lineage_id}",
    headers=HEADERS,
    params={"depth": depth, "direction": "downstream"}
)

if response.status_code == 404:
    raise ValueError(f"Lineage ID '{lineage_id}' not found. Was it registered?")

if response.status_code != 200:
    raise RuntimeError(f"Lineage fetch failed [{response.status_code}]: {response.text}")

return response.json()

def pretty_print_graph(graph: dict) -> None:
"""Print a human-readable dependency tree."""
print(f"\n📊 Lineage Graph: {graph['root_asset']['name']}")
print(f" Total downstream consumers: {graph['stats']['total_consumers']}")
print(f" Pipeline types involved: {', '.join(graph['stats']['pipeline_types'])}\n")

for node in graph["nodes"]:
    depth_indent = "  " * node["depth"]
    status_icon = "⚠️ " if node.get("has_warnings") else "✓ "
    print(f"{depth_indent}{status_icon}{node['name']} ({node['type']})")
    if node.get("owner"):
        print(f"{depth_indent}   owner: {node['owner']}")

graph = get_lineage("lin_7f3a9c2e")
pretty_print_graph(graph)

**Expected output:**

plaintext
📊 Lineage Graph: raw.users
Total downstream consumers: 4
Pipeline types involved: dbt, airflow, custom

✓ marts.user_metrics (dbt_model)
owner: data-team
✓ revenue_report_dag (airflow_dag)
owner: analytics-eng
⚠️ legacy_export_script (custom_etl)
owner: unknown

That warning on `legacy_export_script`? That's DataLineage flagging an asset with no owner and no documentation. That's The Greg Script™ being a liability in graph form.

---

## Step 3: Run Impact Analysis Before a Schema Change

Here's where this pays for itself. Before you `ALTER TABLE users DROP COLUMN legacy_id`, you run this:

python

impact_analysis.py

import requests
from config import BASE_URL, HEADERS

def analyze_impact(asset_name: str, change_type: str, changed_fields: list) -> dict:
"""
Predict blast radius of a proposed schema change.

Args:
    asset_name: The asset being changed
    change_type: 'column_drop', 'column_rename', 'type_change', 'table_drop'
    changed_fields: List of field names being modified

Returns:
    Impact report with affected consumers and severity scores
"""
payload = {
    "asset": {"type": "table", "name": asset_name},
    "proposed_change": {
        "type": change_type,
        "fields": changed_fields
    }
}

response = requests.post(
    f"{BASE_URL}/lineage/impact",
    headers=HEADERS,
    json=payload
)

if response.status_code != 200:
    raise RuntimeError(
        f"Impact analysis failed [{response.status_code}]: "
        f"{response.json().get('message', response.text)}"
    )

return response.json()

def format_impact_report(report: dict) -> None:
"""Format impact report as a pre-change checklist."""
severity_emoji = {"critical": "🔴", "high": "🟠", "medium": "🟡", "low": "🟢"}

print(f"\n{'='*50}")
print(f"IMPACT REPORT: {report['proposed_change']['type'].upper()}")
print(f"{'='*50}")
print(f"Overall risk: {severity_emoji.get(report['risk_level'], '⚪')} {report['risk_level'].upper()}")
print(f"Affected consumers: {report['affected_count']}\n")

print("AFFECTED ASSETS (resolve before deploying):")
for consumer in report["affected_consumers"]:
    emoji = severity_emoji.get(consumer["severity"], "⚪")
    print(f"\n  {emoji} {consumer['name']}")
    print(f"     Type: {consumer['type']}")
    print(f"     Uses field(s): {', '.join(consumer['referenced_fields'])}")
    print(f"     Owner: {consumer.get('owner', 'UNKNOWN — investigate immediately')}")
    print(f"     Action required: {consumer['recommended_action']}")

Simulate: we want to drop the 'legacy_id' column from raw.users

impact = analyze_impact(
asset_name="raw.users",
change_type="column_drop",
changed_fields=["legacy_id"]
)

format_impact_report(impact)

**Expected output:**

plaintext

IMPACT REPORT: COLUMN_DROP

Overall risk: 🔴 CRITICAL
Affected consumers: 3

AFFECTED ASSETS (resolve before deploying):

🟠 marts.user_metrics (dbt_model)
Type: dbt_model
Uses field(s): legacy_id
Owner: data-team
Action required: Update model to remove legacy_id reference

🔴 legacy_export_script (custom_etl)
Type: custom_etl
Uses field(s): legacy_id
Owner: UNKNOWN — investigate immediately
Action required: Locate script owner before proceeding

🟡 revenue_report_dag (airflow_dag)
Type: airflow_dag
Uses field(s): legacy_id (indirect via marts.user_metrics)
Owner: analytics-eng
Action required: Will self-resolve after dbt model update

This output is your pre-flight checklist. Print it. Paste it in your PR description. Send it to The Greg Script™'s owner (whoever that is).

---

## Pulling It Together: A Pre-Change Workflow

Here's the pattern I'd recommend building into your deployment process:

python

pre_change_check.py — run this before any schema migration

from impact_analysis import analyze_impact, format_impact_report

def pre_change_gate(asset_name: str, change_type: str, fields: list) -> bool:
"""
Returns True if safe to proceed, False if blockers exist.
Designed to plug into CI/CD pipelines.
"""
report = analyze_impact(asset_name, change_type, fields)
format_impact_report(report)

blockers = [c for c in report["affected_consumers"] if c["severity"] == "critical"]

if blockers:
    print(f"\n🚫 BLOCKED: {len(blockers)} critical issue(s) must be resolved first.")
    return False

print(f"\n✅ CLEAR TO PROCEED (with caution)")
return True

---

## What This Changes About How You Work

The shift DataLineage enables isn't just technical — it's cultural. Instead of "move fast and find out," your team can operate on "move fast and *know*."

A few practices worth adopting once you have lineage tracking in place:

- **Register assets at deploy time**, not after incidents. Treat it like writing tests — a discipline, not a reaction.
- **Add impact analysis to your migration CI step.** A failed gate beats a 2 AM page.
- **Use the `owner` metadata field religiously.** The Greg Script™ situation is avoidable.

The goal isn't a perfect dependency map. The goal is *enough* visibility that Tuesday morning stays boring.

---

*Have a pipeline horror story? Drop it in the comments — bonus points if it involved a column rename that took down three dashboards.*