Data Pipeline Monitoring: Failures fangen, bevor Stakeholder:innen es tun

import requests
from airflow.models import Variable

def slack_failure_alert(context):
    webhook_url = Variable.get("slack_webhook_url")
    task_instance = context.get("task_instance")
    payload = {
        "text": (
            f":rotating_light: *Pipeline-Failure*\n"
            f"*DAG:* `{task_instance.dag_id}`\n"
            f"*Task:* `{task_instance.task_id}`\n"
            f"*Execution Date:* {context.get('execution_date').isoformat()}\n"
            f"*Logs:* <{task_instance.log_url}|View Logs>"
        )
    }
    requests.post(webhook_url, json=payload, timeout=10)


def sla_miss_alert(dag, task_list, blocking_task_list, slas, blocking_tis):
    webhook_url = Variable.get("slack_webhook_url")
    task_names = ", ".join([t.task_id for t in task_list])
    payload = {
        "text": (
            f":warning: *SLA-Miss*\n"
            f"*DAG:* `{dag.dag_id}`\n"
            f"*Tasks:* `{task_names}`\n"
            f"*Erwartet bis:* {slas[0].execution_date.isoformat()}"
        )
    }
    requests.post(webhook_url, json=payload, timeout=10)

WITH row_check AS (
    SELECT COUNT(*) AS row_count,
        CASE
            WHEN COUNT(*) = 0 THEN 'CRITICAL: Zero rows'
            WHEN COUNT(*) < 1000 THEN 'WARNING: Low'
            WHEN COUNT(*) > 1000000 THEN 'WARNING: High'
            ELSE 'OK'
        END AS status
    FROM analytics.daily_revenue WHERE load_date = CURRENT_DATE
),
null_check AS (
    SELECT
        ROUND(100.0 * COUNT(*) FILTER (WHERE revenue IS NULL) / NULLIF(COUNT(*), 0), 2) AS revenue_null_pct,
        ROUND(100.0 * COUNT(*) FILTER (WHERE customer_id IS NULL) / NULLIF(COUNT(*), 0), 2) AS customer_id_null_pct
    FROM analytics.daily_revenue WHERE load_date = CURRENT_DATE
),
freshness_check AS (
    SELECT MAX(event_timestamp) AS latest_event,
        CASE
            WHEN MAX(event_timestamp) < NOW() - INTERVAL '24 hours' THEN 'CRITICAL: > 24h'
            WHEN MAX(event_timestamp) < NOW() - INTERVAL '6 hours' THEN 'WARNING: > 6h'
            ELSE 'OK'
        END AS freshness_status
    FROM analytics.daily_revenue WHERE load_date = CURRENT_DATE
),
pk_check AS (
    SELECT
        CASE WHEN COUNT(*) != COUNT(DISTINCT transaction_id)
        THEN 'CRITICAL: Duplicate PKs' ELSE 'OK' END AS uniqueness_status
    FROM analytics.daily_revenue WHERE load_date = CURRENT_DATE
)
SELECT r.row_count, r.status, n.revenue_null_pct, n.customer_id_null_pct,
       f.freshness_status, p.uniqueness_status
FROM row_check r, null_check n, freshness_check f, pk_check p;

from sqlalchemy import create_engine, inspect
import json
from pathlib import Path

def get_current_schema(engine, table_name: str, schema: str = "public") -> dict:
    inspector = inspect(engine)
    columns = inspector.get_columns(table_name, schema=schema)
    return { col["name"]: str(col["type"]) for col in columns }

def detect_schema_changes(engine, table_name: str,
                          snapshot_dir: str = "/tmp/schema_snapshots",
                          schema: str = "public") -> dict:
    snapshot_path = Path(snapshot_dir) / f"{schema}_{table_name}.json"
    current = get_current_schema(engine, table_name, schema)
    changes = {"added": [], "removed": [], "type_changed": {}}

    if snapshot_path.exists():
        previous = json.loads(snapshot_path.read_text())
        current_cols = set(current.keys())
        previous_cols = set(previous.keys())
        changes["added"] = sorted(current_cols - previous_cols)
        changes["removed"] = sorted(previous_cols - current_cols)
        for col in current_cols & previous_cols:
            if current[col] != previous[col]:
                changes["type_changed"][col] = {"old": previous[col], "new": current[col]}

    snapshot_path.parent.mkdir(parents=True, exist_ok=True)
    snapshot_path.write_text(json.dumps(current, indent=2))
    return changes