A look at how column-level lineage works in dbt — in open-source, dbt Cloud, and through SQLMesh. There are more differences than you think

When every dashboard refreshes on time and the numbers are still wrong, the problem isn't the model. Continue reading...

One-click tool that shows exactly where fields, measures, and tables are used across every Power BI report - down to the visual. All workspaces, full lineage, and a whole lot more for an all-in-one governance and impact solution.

On 3 August 2026 the Financial Crimes Enforcement Network (FinCEN) imposed a civil money penalty of $125 million on UBS Financial Services Inc (UBSFS), the largest ever levied on a broker-dealer for v...

Power BI Sentinel is a SaaS application that works alongside your Power BI estate, to provide greater security and GDPR data governance of your Power BI data. It is an enterprise-level analytics…

another post form Cybertec

--- title: "Stop Playing Data Detective: Automate Pipeline Dependency Tracing with DataLineage" published: false tags: [dataengineering, python, tutorial, dataquality] --- # Stop Playing Data Detective: Automate Pipeline Dependency Tracing with DataLineage You know the feeling. It's 2pm on a Tuesday. Someone changed a column name in a source table. By 4pm, three dashboards are broken, a Spark job is throwing cryptic NullPointerExceptions, and your Slack DMs look like a crime scene. The culprit isn't the schema change — it's the fact that nobody *knew* what depended on that column. This tutorial walks you through wiring DataLineage into your stack so that the next time a schema shifts, you're the person who already has the answer before anyone asks the question. --- ## What We're Building By the end of this post, you'll have a working Python script that: 1. Traces dependencies across a mixed pipeline (dbt + Airflow + custom ETL) 2. Stores a lineage graph you can query later 3. Runs an impact analysis *before* a schema change ships We'll use three endpoints throughout: - `POST /lineage/trace` — discover and register dependencies - `GET /lineage/{id}` — retrieve a stored lineage graph - `POST /lineage/impact` — simulate the blast radius of a change --- ## Prerequisites bash pip install requests python-dotenv Set up your environment: bash # .env DATALINEAGE_API_KEY=your_api_key_here DATALINEAGE_BASE_URL=https://api.datalineage.io/v1 --- ## Step 1: Build Your API Client Before touching any pipeline logic, let's build a thin client wrapper. This keeps auth and error handling in one place — a pattern you'll thank yourself for at 2am. python # lineage_client.py import os import requests from dotenv import load_dotenv load_dotenv() class LineageClient: 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 response handling with meaningful error messages.""" 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 DATALINEAGE_BASE_URL." ) def trace(self, payload: dict) -> dict: response = requests.post( f"{self.base_url}/lineage/trace", json=payload, headers=self.headers, timeout=30 ) 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=15 ) return self._handle_response(response) def impact(self, payload: dict) -> dict: response = requests.post( f"{self.base_url}/lineage/impact", json=payload, headers=self.headers, timeout=30 ) return self._handle_response(response) --- ## Step 2: Trace Your First Pipeline Now let's register a real dependency graph. The payload describes your pipeline topology — what tools are involved and which assets connect them. python # trace_pipeline.py from lineage_client import LineageClient client = LineageClient() pipeline_payload = { "pipeline_name": "customer_revenue_pipeline", "environment": "production", "sources": [ { "tool": "custom_etl", "asset": "raw.stripe_events", "schema": { "customer_id": "string", "amount_cents": "integer", "event_timestamp": "timestamp" } } ], "transformations": [ { "tool": "dbt", "model": "stg_stripe_events", "depends_on": ["raw.stripe_events"], "columns_used": ["customer_id", "amount_cents", "event_timestamp"] }, { "tool": "dbt", "model": "fct_customer_revenue", "depends_on": ["stg_stripe_events"] } ], "consumers": [ { "tool": "airflow", "dag_id": "revenue_reporting_dag", "depends_on": ["fct_customer_revenue"] }, { "tool": "spark", "job_name": "ml_feature_extraction", "depends_on": ["fct_customer_revenue"], "columns_used": ["customer_id", "amount_cents"] } ] } try: result = client.trace(pipeline_payload) lineage_id = result["lineage_id"] print(f"✅ Lineage registered successfully.") print(f" Lineage ID: {lineage_id}") print(f" Nodes discovered: {result['node_count']}") print(f" Edges mapped: {result['edge_count']}") except RuntimeError as e: print(f"❌ Trace failed: {e}") **Expected output:** plaintext ✅ Lineage registered successfully. Lineage ID: lin_7f3a9c2e Nodes discovered: 5 Edges mapped: 6 Save that `lineage_id`. It's your handle to everything downstream. --- ## Step 3: Inspect the Dependency Graph Got your ID? Let's pull the full graph and make it human-readable. python # inspect_lineage.py from lineage_client import LineageClient client = LineageClient() LINEAGE_ID = "lin_7f3a9c2e" # from Step 2 def print_dependency_tree(lineage: dict): """Render a simple ASCII dependency tree from the lineage graph.""" nodes = {n["id"]: n for n in lineage["nodes"]} edges = lineage["edges"] # list of {"from": id, "to": id} # Find root nodes (no incoming edges) targets = {e["to"] for e in edges} roots = [n for n in nodes if n not in targets] def render(node_id, depth=0): node = nodes[node_id] prefix = " " * depth + ("└─ " if depth > 0 else "") print(f"{prefix}[{node['tool']}] {node['asset_name']}") children = [e["to"] for e in edges if e["from"] == node_id] for child in children: render(child, depth + 1) print("\n📊 Dependency Tree:") print("=" * 40) for root in roots: render(root) try: lineage = client.get_lineage(LINEAGE_ID) print_dependency_tree(lineage) print(f"\n Last updated: {lineage['updated_at']}") print(f" Health status: {lineage['health_status']}") except RuntimeError as e: print(f"❌ Could not retrieve lineage: {e}") **Expected output:** plaintext # 📊 Dependency Tree: [custom_etl] raw.stripe_events └─ [dbt] stg_stripe_events └─ [dbt] fct_customer_revenue └─ [airflow] revenue_reporting_dag └─ [spark] ml_feature_extraction Last updated: 2024-11-12T14:32:01Z Health status: healthy This is the map that saves you on Tuesday afternoons. --- ## Step 4: Run Impact Analysis Before Shipping a Change Here's where DataLineage earns its keep. Before you rename `amount_cents` to `amount_usd` in your source schema, ask the API what breaks. python # impact_analysis.py from lineage_client import LineageClient client = LineageClient() proposed_change = { "lineage_id": "lin_7f3a9c2e", "change_type": "column_rename", "target_asset": "raw.stripe_events", "change_details": { "column": "amount_cents", "rename_to": "amount_usd" } } def render_impact_report(impact: dict): severity_icons = {"high": "🔴", "medium": "🟡", "low": "🟢"} print("\n⚡ Impact Analysis Report") print("=" * 40) print(f"Proposed change: {impact['change_summary']}") print(f"Affected assets: {impact['total_affected']}\n") for affected in impact["affected_assets"]: icon = severity_icons.get(affected["severity"], "⚪") print(f"{icon} [{affected['tool']}] {affected['asset_name']}") print(f" Reason: {affected['reason']}") print(f" Columns at risk: {', '.join(affected['columns_at_risk'])}") print() if impact["total_affected"] == 0: print("✅ No downstream consumers affected. Safe to ship.") try: impact = client.impact(proposed_change) render_impact_report(impact) except RuntimeError as e: print(f"❌ Impact analysis failed: {e}") **Expected output:** plaintext # ⚡ Impact Analysis Report Proposed change: Rename 'amount_cents' → 'amount_usd' in raw.stripe_events Affected assets: 2 🔴 [spark] ml_feature_extraction Reason: Directly references column 'amount_cents' Columns at risk: amount_cents 🟡 [dbt] stg_stripe_events Reason: Selects all columns from source; may inherit rename Columns at risk: amount_cents Two affected assets, surfaced in seconds. Before you type a single migration script. --- ## Putting It All Together: A Pre-Deployment Hook Combine everything into a script you can run in CI before any schema migration merges: python # pre_deploy_check.py import sys from lineage_client import LineageClient def run_pre_deploy_check(lineage_id: str, change_payload: dict) -> bool: client = LineageClient() print(f"🔍 Running impact analysis for lineage: {lineage_id}") try: impact = client.impact({"lineage_id": lineage_id, **change_payload}) high_severity = [ a for a in impact["affected_assets"] if a["severity"] == "high" ] if high_severity: print(f"🚫 Deployment blocked: {len(high_severity)} high-severity impact(s) detected.") for asset in high_severity: print(f" - [{asset['tool']}] {asset['asset_name']}: {asset['reason']}") return False print(f"✅ No high-severity impacts. Deployment cleared.") return True except RuntimeError as e: print(f"⚠️ Could not complete impact analysis: {e}") print(" Blocking deployment as a precaution.") return False if **name** == "**main** ": cleared = run_pre_deploy_check( lineage_id="lin_7f3a9c2e", change_payload={ "change_type": "column_rename", "target_asset": "raw.stripe_events", "change_details": {"column": "amount_cents", "rename_to": "amount_usd"} } ) sys.exit(0 if cleared else 1) Drop this in your CI pipeline. Schema changes that break downstream consumers never reach production again. --- ## What's Next You've got the foundation. From here, consider: - **Scheduling `trace` calls** in Airflow after each pipeline run to keep lineage fresh - **Alerting on `health_status` changes** from `GET /lineage/{id}` — catch drift before users do - **Extending the impact payload** with `change_type: "column_drop"` for deletion risk analysis The goal isn't just knowing what broke — it's building a system where you know *before* it breaks. That's the difference between being reactive and being the engineer everyone trusts to ship safely. Now go enjoy your Tuesday afternoons.

Sanjay K Mohindroo

Discover how data governance shapes business success. Real stories, best practices, and debate on data quality and security. #DataGovernance #DataQuality #DataSecurity #DataStewardship #DataManagementBestPractices #DataCompliance #DataOwnership #DataLineage #DataCulture #DataDrivenDecisions #DataAudit #DataTrust #DataProtection #DataStewardshipTeam #DataDictionary

Discover how data governance shapes business success. Real stories, best practices, and debate on data quality and security. #DataGovernance #DataQuality #DataSecurity #DataStewardship #DataManagementBestPractices #DataCompliance #DataOwnership #DataLineage #DataCulture #DataDrivenDecisions #DataAudit #DataTrust #DataProtection #DataStewardshipTeam #DataDictionary

AI insights shouldn’t live in a vacuum. Learn how enterprises can govern AI data to reduce risk and improve ROI. Continue reading...

Without data lineage, you don't really know your data. Continue reading...

CP4D4.6.4でのWKCのビジネスデータリネージュとMANTAリネージュのデモ動画です。2分程度。