How to Track Data Pipeline Dependencies Automatically with DataLineage

---
title: "Stop Playing Detective: Automate Data Lineage Tracing Across Your Entire Pipeline Stack"
published: false
tags: [dataengineering, python, tutorial, dataquality]
---

# Stop Playing Detective: Automate Data Lineage Tracing Across Your Entire Pipeline Stack

You know the scene. A Slack message arrives at 9 AM: *"The revenue dashboard looks wrong."* You open your laptop and begin the archaeological dig — tracing data backward through Airflow DAGs, dbt models, Spark jobs, and three custom ETL scripts written by someone who left the company in 2022. Two hours later, you've found the culprit: a schema change in an upstream table that nobody knew had seventeen downstream consumers.

This tutorial will show you how to use **DataLineage** to make that investigation take thirty seconds instead of two hours.

---

## What We're Building

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

1. Registers your pipeline assets with DataLineage
2. Traces dependencies between them automatically
3. Runs an impact analysis before any schema change ships

We'll simulate a realistic stack: an Airflow-ingested raw table, a dbt model that transforms it, and a Spark job that feeds a downstream dashboard.

---

## Prerequisites

bash
pip install requests python-dotenv

Set your API key in a `.env` file:

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

---

## Step 1: Build a Reusable Client

Before touching the API, let's write a thin client wrapper. This keeps authentication and error handling in one place — a pattern your future self will thank you for.

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:
    try:
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        error_body = response.json() if response.content else {}
        raise RuntimeError(
            f"API error {response.status_code}: "
            f"{error_body.get('message', str(e))}"
        ) from e
    except requests.exceptions.ConnectionError:
        raise RuntimeError("Could not reach DataLineage API. Check your network and base URL.")

def trace(self, payload: dict) -> dict:
    response = requests.post(
        f"{self.base_url}/lineage/trace",
        json=payload,
        headers=self.headers,
        timeout=10,
    )
    return self._handle_response(response)

def get_lineage(self, lineage_id: str) -> dict:
    response = requests.get(
        f"{self.base_url}/lineage/{lineage_id}",
        headers=self.headers,
        timeout=10,
    )
    return self._handle_response(response)

def impact_analysis(self, payload: dict) -> dict:
    response = requests.post(
        f"{self.base_url}/lineage/impact",
        json=payload,
        headers=self.headers,
        timeout=10,
    )
    return self._handle_response(response)

client = DataLineageClient()
print("Client initialized successfully.")

**Expected output:**

plaintext
Client initialized successfully.

---

## Step 2: Register Your Pipeline Assets

Now let's tell DataLineage about our pipeline. We'll model three assets that represent a real-world ingestion → transformation → consumption chain.

python
pipeline_assets = [
{
"asset_id": "raw.orders",
"asset_type": "table",
"tool": "airflow",
"schema": {
"columns": ["order_id", "customer_id", "amount", "created_at"],
},
"description": "Raw orders ingested from the transactional DB via Airflow.",
},
{
"asset_id": "dbt.orders_daily",
"asset_type": "model",
"tool": "dbt",
"schema": {
"columns": ["order_date", "customer_id", "total_amount", "order_count"],
},
"description": "Daily order aggregation built on raw.orders.",
"upstream": ["raw.orders"],
},
{
"asset_id": "spark.revenue_summary",
"asset_type": "dataset",
"tool": "spark",
"schema": {
"columns": ["week", "region", "revenue"],
},
"description": "Weekly revenue rollup consumed by the executive dashboard.",
"upstream": ["dbt.orders_daily"],
},
]

trace_result = client.trace({
"assets": pipeline_assets,
"auto_discover": True,
})

lineage_id = trace_result["lineage_id"]
print(f"Lineage graph registered. ID: {lineage_id}")
print(f"Assets discovered: {trace_result['asset_count']}")
print(f"Edges mapped: {trace_result['edge_count']}")

**Expected output:**

plaintext
Lineage graph registered. ID: lin_8f3a92bc
Assets discovered: 3
Edges mapped: 2

The `auto_discover: true` flag tells DataLineage to also scan your connected integrations for any assets it can infer automatically — useful when you have undocumented dependencies lurking in legacy scripts.

---

## Step 3: Inspect the Full Lineage Graph

With the graph registered, let's pull it back and walk the dependency tree programmatically.

python
import json

lineage = client.get_lineage(lineage_id)

print("\n=== Lineage Graph ===\n")
for node in lineage["nodes"]:
upstream_list = ", ".join(node.get("upstream", [])) or "None (source)"
print(f" [{node['tool'].upper()}] {node['asset_id']}")
print(f" └─ Upstream: {upstream_list}")
print(f" └─ Columns: {', '.join(node['schema']['columns'])}\n")

**Expected output:**

plaintext
=== Lineage Graph ===

[AIRFLOW] raw.orders
└─ Upstream: None (source)
└─ Columns: order_id, customer_id, amount, created_at

[DBT] dbt.orders_daily
└─ Upstream: raw.orders
└─ Columns: order_date, customer_id, total_amount, order_count

[SPARK] spark.revenue_summary
└─ Upstream: dbt.orders_daily
└─ Columns: week, region, revenue

This is the map you'd normally reconstruct manually from scattered YAML files and Slack threads.

---

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

Here's where DataLineage earns its keep. Before you rename `amount` to `order_amount` in `raw.orders`, let's see exactly what breaks.

python
proposed_change = {
"asset_id": "raw.orders",
"change_type": "column_rename",
"details": {
"old_name": "amount",
"new_name": "order_amount",
},
"lineage_id": lineage_id,
}

impact = client.impact_analysis(proposed_change)

print("\n=== Impact Analysis Report ===\n")
print(f"Change: rename '{proposed_change['details']['old_name']}' "
f"→ '{proposed_change['details']['new_name']}' "
f"in {proposed_change['asset_id']}\n")

if not impact["affected_assets"]:
print("✅ No downstream consumers affected. Safe to proceed.")
else:
print(f"⚠️ {len(impact['affected_assets'])} downstream asset(s) affected:\n")
for asset in impact["affected_assets"]:
severity = asset.get("severity", "unknown").upper()
print(f" [{severity}] {asset['asset_id']} ({asset['tool']})")
print(f" Reason: {asset['reason']}")
print(f" Suggested fix: {asset.get('suggested_fix', 'Manual review required')}\n")

**Expected output:**

plaintext
=== Impact Analysis Report ===

Change: rename 'amount' → 'order_amount' in raw.orders

⚠️ 2 downstream asset(s) affected:

[HIGH] dbt.orders_daily (dbt)
Reason: References column 'amount' in aggregation logic.
Suggested fix: Update SUM(amount) to SUM(order_amount) in orders_daily.sql

[HIGH] spark.revenue_summary (spark)
Reason: Inherits 'amount' column through dbt.orders_daily transformation.
Suggested fix: Re-run dbt.orders_daily after fix; verify Spark job schema mapping.

Two assets. Flagged. With suggested fixes. In under a second.

---

## Step 5: Wire It Into Your CI/CD Pipeline

The real power comes from making this check automatic. Here's a minimal GitHub Actions step that blocks a PR if an impact analysis returns high-severity hits:

python

impact_gate.py — run this in CI before schema migrations ship

import sys

def check_impact_gate(lineage_id: str, asset_id: str, change: dict) -> None:
impact = client.impact_analysis({
"asset_id": asset_id,
"lineage_id": lineage_id,
**change,
})

high_severity = [
    a for a in impact.get("affected_assets", [])
    if a.get("severity") == "high"
]

if high_severity:
    print(f"❌ CI gate failed: {len(high_severity)} high-severity impact(s) detected.")
    for asset in high_severity:
        print(f"   - {asset['asset_id']}: {asset['reason']}")
    sys.exit(1)  # Fail the pipeline
else:
    print("✅ Impact gate passed. No high-severity consumers affected.")
    sys.exit(0)

check_impact_gate(
lineage_id=lineage_id,
asset_id="raw.orders",
change={"change_type": "column_rename", "details": {"old_name": "amount", "new_name": "order_amount"}},
)

Drop this into a CI step with `python impact_gate.py` and schema changes can never silently break downstream consumers again.

---

## What You've Built

In about fifty lines of Python, you've replaced a two-hour manual investigation with an automated, CI-integrated lineage system that:

- **Maps** your full dbt + Airflow + Spark dependency graph
- **Surfaces** every downstream consumer before a breaking change ships
- **Blocks** high-risk schema changes at the PR level

The archaeological digs are over. Your 9 AM Slack messages just got a lot less stressful.

---

*Have a multi-tool pipeline with custom ETL? The `POST /lineage/trace` endpoint accepts a `custom_etl` asset type — drop your questions in the comments.*