Data Lineage

Data Lineage im Überblick

Data Lineage dokumentiert den Weg deiner Daten: Woher kommen sie, was passiert mit ihnen, und wo landen sie am Ende. Wie ein GPS-Tracker für Daten.

Beispiel:

Quelle                    Transformation              Ziel
───────────────────────────────────────────────────────────
CRM-System ──┐
             ├──→ ETL-Job ──→ Data Warehouse ──→ ML-Modell
Web-Logs ────┘      │              │
                    │              └──→ Dashboard
                    │
                    └──→ Aggregation ──→ Report

Was wird getrackt?

Technisch betrachtet

dbt Lineage

-- models/staging/stg_orders.sql
SELECT
    id,
    customer_id,
    order_date,
    total
FROM {{ source('raw', 'orders') }}
WHERE order_date >= '{{ start_date }}'

-- models/marts/customer_metrics.sql
SELECT
    customer_id,
    COUNT(*) as order_count,
    SUM(total) as lifetime_value
FROM {{ ref('stg_orders') }}
GROUP BY customer_id

Transformations-Tools können daraus Lineage ableiten:

raw.orders → stg_orders → customer_metrics

OpenLineage Event

{
  "eventType": "COMPLETE",
  "eventTime": "<event-timestamp>",
  "run": {
    "runId": "abc-123"
  },
  "job": {
    "namespace": "etl",
    "name": "transform_orders"
  },
  "inputs": [
    {
      "namespace": "postgres",
      "name": "raw.orders"
    }
  ],
  "outputs": [
    {
      "namespace": "warehouse",
      "name": "staging.orders"
    }
  ]
}

Airflow mit OpenLineage

from airflow import DAG
from airflow.providers.openlineage.plugins.adapter import OpenLineageAdapter

with DAG("etl_pipeline") as dag:
    
    extract = PostgresOperator(
        task_id="extract_orders",
        sql="SELECT * FROM orders"
    )
    
    transform = PythonOperator(
        task_id="transform",
        python_callable=transform_data
    )
    
    load = BigQueryOperator(
        task_id="load_to_bq",
        destination_table="warehouse.orders"
    )
    
    extract >> transform >> load

# Lineage-Metadaten können erfassen:
# source.orders → transform → warehouse.orders

Impact Analysis

def get_downstream_dependencies(table_name):
    """Was bricht, wenn ich diese Tabelle ändere?"""
    lineage = get_lineage_graph()
    
    downstream = []
    queue = [table_name]
    
    while queue:
        current = queue.pop(0)
        for dependent in lineage.get_dependents(current):
            downstream.append(dependent)
            queue.append(dependent.name)
    
    return downstream

# Beispiel
deps = get_downstream_dependencies("raw.orders")
# → abhängige Tabellen, Dashboards oder Modelle

Lineage für ML

Training Data Lineage:
──────────────────────
raw.customers ──┐
                ├──→ feature_engineering ──→ training_data ──→ model_version
raw.transactions┘         │
                          └──→ feature_store
                          
Wenn Quelldaten sich ändern:
→ Feature Engineering prüfen
→ Training Data neu validieren
→ Modellqualität und Retraining-Bedarf bewerten