Data Lineage implementieren: Schritt-fuer-Schritt-Guide

Beispiel: Python-Script fuer 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()

Beispiel: Airflow DAG fuer 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

Wie implementiert man Data Lineage?

Data Lineage implementierst du mit: 1) Scope und Ziele definieren, 2) Tools waehlen (Catalog, Open Source, manuell), 3) Architektur planen (Metadata-Speicher, Ingestion, Visualisierung), 4) Automatische Erfassung (Query-Logs, APIs, Manifeste), 5) Stack anbinden (Warehouse, ETL, BI), 6) Genauigkeit pruefen, 7) Governance-Workflows aktivieren, 8) Nutzer trainieren. Timeline: 1-3 Monate mit modernen Tools, 3-6 Monate bei Enterprise-Plattformen.

Wie automatisiert man Data Lineage am besten?

Beste Automation: 1) Query-Logs fuer Warehouses (Snowflake, BigQuery, Redshift) parsen, 2) Metadata-APIs aus BI-Tools (Looker, Tableau) via REST holen, 3) dbt-Manifest (manifest.json) einlesen, 4) OpenLineage in Airflow/Spark fuer Runtime-Events. Tools wie Atlan oder Select Star machen all das out-of-the-box.

Wie lange dauert die Implementierung?

Mit modernen Cloud-Catalogs (Atlan, Select Star) 1-4 Wochen Start. Enterprise-Plattformen (Informatica, Collibra) 2-6 Monate. Open Source (DataHub, OpenLineage) 1-3 Monate Engineering. Visuelle Doku-Tools (Datadef) liefern sofort Ergebnisse, brauchen aber manuelle Pflege.

Welche Tools braucht man?

Kern-Tools: 1) Lineage-Plattform (Atlan, Collibra oder DataHub), 2) Metadata-Extractor fuer deinen Stack (dbt, Airflow, Looker), 3) Query-Log-Zugriff (Snowflake ACCOUNT_USAGE, BigQuery INFORMATION_SCHEMA), 4) Orchestrierung fuer geplante Refreshes, 5) Optional: OpenLineage, visuelle Doku-Tools wie Datadef.