How to Implement Data Lineage: Step-by-Step Guide

Example: Python Script to Extract Snowflake Lineage

import snowflake.connector
import json
from collections import defaultdict

# Connect to Snowflake
conn = snowflake.connector.connect(
    account='YOUR_ACCOUNT',
    user='YOUR_USER',
    password='YOUR_PASSWORD',
    warehouse='COMPUTE_WH',
    database='SNOWFLAKE',
    schema='ACCOUNT_USAGE'
)

# Query recent queries for lineage
query = """
SELECT 
    query_id,
    query_text,
    database_name,
    schema_name,
    user_name,
    start_time
FROM snowflake.account_usage.query_history
WHERE query_type IN ('INSERT', 'MERGE', 'CREATE_TABLE_AS_SELECT')
  AND start_time >= DATEADD(day, -7, CURRENT_TIMESTAMP())
ORDER BY start_time DESC
LIMIT 1000;
"""

cursor = conn.cursor()
cursor.execute(query)

# Parse queries and build lineage
lineage_graph = defaultdict(list)

for row in cursor:
    query_id, query_text, db, schema, user, timestamp = row
    
    # Simple parsing (use sqlparse or sqlglot for production)
    if 'INSERT INTO' in query_text.upper():
        # Extract target table
        target = extract_table_name(query_text, 'INSERT INTO')
        # Extract source tables from FROM/JOIN clauses
        sources = extract_table_names(query_text, ['FROM', 'JOIN'])
        
        for source in sources:
            lineage_graph[source].append({
                'target': target,
                'query_id': query_id,
                'user': user,
                'timestamp': str(timestamp)
            })

# Output lineage as JSON
with open('lineage_output.json', 'w') as f:
    json.dump(dict(lineage_graph), f, indent=2)

print(f"Extracted lineage for {len(lineage_graph)} source tables")
conn.close()

Example: Airflow DAG for Scheduled Lineage Extraction

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import requests

def extract_snowflake_lineage():
    """Extract lineage from Snowflake and send to catalog"""
    # Your lineage extraction logic here
    lineage_data = get_snowflake_lineage()
    
    # Upload to catalog tool API
    response = requests.post(
        'https://catalog.example.com/api/lineage',
        headers={'Authorization': f'Bearer {CATALOG_API_KEY}'},
        json=lineage_data
    )
    response.raise_for_status()
    print(f"Uploaded {len(lineage_data)} lineage edges")

def extract_dbt_lineage():
    """Parse dbt manifest and upload"""
    with open('/dbt/target/manifest.json') as f:
        manifest = json.load(f)
    
    # Upload to catalog
    response = requests.post(
        'https://catalog.example.com/api/dbt/manifest',
        headers={'Authorization': f'Bearer {CATALOG_API_KEY}'},
        json=manifest
    )
    response.raise_for_status()

default_args = {
    'owner': 'data-platform',
    'depends_on_past': False,
    'start_date': datetime(2025, 1, 1),
    'email_on_failure': True,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'lineage_extraction',
    default_args=default_args,
    description='Daily lineage metadata extraction',
    schedule_interval='@daily',
    catchup=False,
) as dag:

    extract_snowflake = PythonOperator(
        task_id='extract_snowflake_lineage',
        python_callable=extract_snowflake_lineage,
    )

    extract_dbt = PythonOperator(
        task_id='extract_dbt_lineage',
        python_callable=extract_dbt_lineage,
    )

    extract_snowflake >> extract_dbt

How do you implement data lineage?

To implement data lineage: 1) Define scope and objectives, 2) Choose tools (automated catalog, open source, or manual), 3) Design architecture (metadata storage, ingestion, visualization), 4) Set up automated capture (query logs, APIs, manifests), 5) Integrate with data stack (warehouses, ETL, BI), 6) Validate accuracy, 7) Enable governance workflows, 8) Train users. Timeline: 1-3 months for modern tools, 3-6 months for enterprise platforms.

What is the best way to automate data lineage?

Best automation methods: 1) Query log parsing for warehouses (Snowflake, BigQuery, Redshift) extracts SQL to infer lineage, 2) Metadata API extraction from BI tools (Looker, Tableau) via REST APIs, 3) Manifest ingestion from dbt (manifest.json parsing), 4) OpenLineage instrumentation in Airflow/Spark for runtime events. Modern tools like Atlan and Select Star automate all of these methods.

How long does it take to implement data lineage?

Implementation timeline varies by approach: Modern cloud catalogs (Atlan, Select Star) take 1-4 weeks for initial setup. Enterprise platforms (Informatica, Collibra) require 2-6 months with consulting. Open-source tools (DataHub, OpenLineage) need 1-3 months with engineering resources. Visual documentation tools (Datadef) provide instant results but require manual maintenance.

What tools are needed to implement data lineage?

Essential tools: 1) Lineage platform (Atlan, Collibra, or DataHub), 2) Metadata extractors for your stack (dbt, Airflow, Looker connectors), 3) Query log access to warehouses (Snowflake ACCOUNT_USAGE, BigQuery INFORMATION_SCHEMA), 4) Orchestration for scheduled metadata refresh, 5) Optional: OpenLineage for vendor-neutral instrumentation, visual tools like Datadef for documentation.