How to Track Data Pipeline Dependencies Automatically with DataLineage

---
title: "Stop Playing Data Detective: Let DataLineage Trace Your Pipeline Dependencies"
published: false
tags: [dataengineering, python, tutorial, dbt]
---

# Stop Playing Data Detective: Let DataLineage Trace Your Pipeline Dependencies

Picture this: it's 3pm on a Friday. Someone renamed a column in a source table. By Monday morning, three dashboards are broken, one Airflow DAG is throwing cryptic errors, and your Spark job silently swallowed bad nulls for 48 hours before anyone noticed.

You've been there. We've all been there.

The problem isn't that pipelines break — it's that when they break, you're essentially an archaeologist. You dig through dbt docs, grep through DAG definitions, and DM the person who "probably knows" what consumes that table. It's manual, it's slow, and it scales terribly.

This tutorial walks you through DataLineage — a tool that automatically maps dependencies across your entire stack (dbt, Airflow, Spark, custom ETL) so that when something changes, you know *exactly* what's downstream before it bites you.

Let's build something real.

---

## What We're Building

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

1. Traces the full dependency graph for any dataset
2. Runs an impact analysis *before* a schema change ships
3. Handles errors gracefully so your CI pipeline doesn't silently lie to you

---

## Prerequisites

bash
pip install requests python-dotenv

Grab your API key from the DataLineage dashboard and store it safely:

bash

.env

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

---

## Step 1: Trace a Dataset's Dependency Graph

The first thing you'll want to do is ask DataLineage: *"who touches this dataset, and what does it touch?"*

We hit `POST /lineage/trace` with the dataset identifier — this can be a table name, a dbt model, or a Spark output path.

python
import os
import requests
from dotenv import load_dotenv

load_dotenv()

BASE_URL = os.getenv("DATALINEAGE_BASE_URL")
HEADERS = {
"Authorization": f"Bearer {os.getenv('DATALINEAGE_API_KEY')}",
"Content-Type": "application/json",
}

def trace_lineage(dataset_id: str, depth: int = 3) -> dict:
"""
Initiate a lineage trace for a given dataset.

Args:
    dataset_id: The unique identifier for your dataset 
                (e.g., 'warehouse.analytics.user_events')
    depth: How many hops upstream/downstream to traverse

Returns:
    The trace job response containing a lineage ID
"""
payload = {
    "dataset_id": dataset_id,
    "depth": depth,
    "include_upstream": True,
    "include_downstream": True,
}

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

# Don't silently swallow HTTP errors — surface them immediately
response.raise_for_status()

return response.json()

--- Run it ---

if name == "main":
result = trace_lineage("warehouse.analytics.user_events")
print(f"Trace initiated. Lineage ID: {result['lineage_id']}")
print(f"Status: {result['status']}")

**Expected output:**

plaintext
Trace initiated. Lineage ID: lin_8f3a92bc
Status: processing

> **Best practice:** Store `lineage_id` — you'll need it to fetch results and run impact analysis. Consider logging it to your observability stack alongside any schema migration you run.

---

## Step 2: Fetch the Full Lineage Graph

Tracing is async (these graphs can be large), so we poll `GET /lineage/{id}` until the job completes. Here's a polling wrapper that won't hammer the API:

python
import time

def fetch_lineage(lineage_id: str, poll_interval: int = 2, max_wait: int = 60) -> dict:
"""
Poll for lineage results until complete or timeout.

Args:
    lineage_id: The ID returned from trace_lineage()
    poll_interval: Seconds between polls
    max_wait: Maximum seconds to wait before giving up

Returns:
    The complete lineage graph

Raises:
    TimeoutError: If the trace doesn't complete within max_wait
    RuntimeError: If the trace job failed on the server side
"""
elapsed = 0

while elapsed < max_wait:
    response = requests.get(
        f"{BASE_URL}/lineage/{lineage_id}",
        headers=HEADERS,
        timeout=30,
    )
    response.raise_for_status()
    data = response.json()

    if data["status"] == "complete":
        return data

    if data["status"] == "failed":
        raise RuntimeError(
            f"Lineage trace failed: {data.get('error', 'Unknown error')}"
        )

    print(f"  Still processing... ({elapsed}s elapsed)")
    time.sleep(poll_interval)
    elapsed += poll_interval

raise TimeoutError(f"Lineage trace {lineage_id} did not complete within {max_wait}s")

def summarize_lineage(lineage_data: dict) -> None:
"""Print a human-readable summary of the lineage graph."""
nodes = lineage_data.get("nodes", [])
edges = lineage_data.get("edges", [])

print(f"\n{'='*50}")
print(f"Lineage Graph: {lineage_data['dataset_id']}")
print(f"{'='*50}")
print(f"Total nodes: {len(nodes)}")
print(f"Total dependencies: {len(edges)}")

print("\nDownstream consumers:")
for node in nodes:
    if node["direction"] == "downstream":
        print(f"  [{node['tool']}] {node['name']}")

--- Run it ---

if name == "main":
trace = trace_lineage("warehouse.analytics.user_events")
lineage = fetch_lineage(trace["lineage_id"])
summarize_lineage(lineage)

**Expected output:**

plaintext
Still processing... (2s elapsed)

==================================================

Lineage Graph: warehouse.analytics.user_events

Total nodes: 14
Total dependencies: 19

Downstream consumers:
[dbt] model.analytics.weekly_retention
[dbt] model.analytics.revenue_attribution

[airflow] dag.nightly_user_export
[spark] job.ml_feature_pipeline
[custom_etl] sync.salesforce_enrichment

Fourteen nodes. Nineteen edges. From a single table. This is exactly why manual tracing doesn't scale.

---

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

This is where DataLineage earns its keep. Before you rename that column, you ask: *"what breaks?"*

`POST /lineage/impact` takes your proposed change and returns a severity-ranked list of affected consumers.

python
def analyze_impact(dataset_id: str, proposed_changes: list[dict]) -> dict:
"""
Assess the downstream impact of proposed schema changes.

Args:
    dataset_id: The dataset you're planning to modify
    proposed_changes: List of change descriptors
        Each dict should have: 'type' (rename/drop/type_change), 
        'field', and optionally 'new_name' or 'new_type'

Returns:
    Impact report with affected consumers ranked by severity
"""
payload = {
    "dataset_id": dataset_id,
    "changes": proposed_changes,
}

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

def print_impact_report(report: dict) -> None:
"""Display a color-coded impact report (terminal-friendly)."""
severity_icons = {
"critical": "🔴",
"high": "🟠",
"medium": "🟡",
"low": "🟢",
}

affected = report.get("affected_consumers", [])

print(f"\n{'='*50}")
print(f"Impact Analysis Report")
print(f"{'='*50}")
print(f"Total affected consumers: {len(affected)}")

if not affected:
    print("✅ No downstream consumers affected. Safe to proceed.")
    return

for consumer in sorted(affected, key=lambda x: x["severity_rank"]):
    icon = severity_icons.get(consumer["severity"], "⚪")
    print(f"\n{icon} {consumer['name']}")
    print(f"   Tool: {consumer['tool']}")
    print(f"   Affected fields: {', '.join(consumer['affected_fields'])}")
    print(f"   Owner: {consumer.get('owner', 'Unknown')}")
    print(f"   Reason: {consumer['reason']}")

--- Run it ---

if name == "main":
changes = [
{
"type": "rename",
"field": "user_id",
"new_name": "customer_id",
},
{
"type": "drop",
"field": "legacy_segment",
}
]

report = analyze_impact("warehouse.analytics.user_events", changes)
print_impact_report(report)

**Expected output:**

plaintext

Impact Analysis Report

Total affected consumers: 6

🔴 model.analytics.revenue_attribution
Tool: dbt
Affected fields: user_id, legacy_segment
Owner: [e-mail]
Reason: Direct field reference in JOIN condition

🟠 dag.nightly_user_export
Tool: airflow
Affected fields: user_id
Owner: [e-mail]
Reason: Field used in SELECT and WHERE clause

🟡 job.ml_feature_pipeline
Tool: spark
Affected fields: legacy_segment
Owner: [e-mail]
Reason: Feature derived from legacy_segment

Now you have a contact list and a blast radius — before you've touched a single line of SQL.

---

## Putting It All Together

Here's a minimal CLI wrapper you can drop into your CI/CD pipeline as a pre-migration gate:

python
import sys

def pre_migration_check(dataset_id: str, changes: list[dict]) -> int:
"""
Returns exit code 0 if safe, 1 if critical impacts found.
Designed for use in CI pipelines.
"""
print(f"Running pre-migration check for: {dataset_id}")

report = analyze_impact(dataset_id, changes)
print_impact_report(report)

critical_count = sum(
    1 for c in report.get("affected_consumers", [])
    if c["severity"] == "critical"
)

if critical_count > 0:
    print(f"\n❌ {critical_count} critical impact(s) found. Resolve before proceeding.")
    return 1

print("\n✅ No critical impacts. Proceeding with migration.")
return 0

if name == "main":
exit_code = pre_migration_check(
dataset_id="warehouse.analytics.user_events",
changes=[{"type": "rename", "field": "user_id", "new_name": "customer_id"}]
)
sys.exit(exit_code)

Add this to your GitHub Actions workflow before any migration step runs, and you've turned a Friday-afternoon fire drill into a boring, automated checklist item.

---

## Key Takeaways

- **Trace first, change second.** Running `POST /lineage/trace` before any schema work costs you 30 seconds and can save you hours.
- **Impact analysis is a contract.** Share the report with consumer team owners before you ship — it's a conversation starter, not just a debugging tool.
- **Automate the gate.** The pre-migration check pattern above is trivially embeddable in any CI system. Make it a required check.
- **Depth matters.** The `depth` parameter on trace is worth tuning — a depth of 2 might miss a dbt model that feeds a Spark job that feeds a dashboard.

Data dependencies are invisible until they're not. DataLineage makes them visible on purpose, before something goes wrong.

Now go rename that column — you've earned it.