English version

Workflows & Orchestrierung

Diese Lektion verdrahtet alles miteinander: Ingestion, Silver, Features, ML-Inferenz und Gold — orchestriert als ein einziger produktiver Workflow. Am Ende weißt du, wie du Task-Abhängigkeiten strukturierst, Fehler sauber abfängst, Retries konfigurierst und das Alerting für deine Wetter-zu-Strom-Pipeline aufsetzt.

Die komplette Pipeline als Workflow

Workflow-Konfiguration (JSON)

Databricks Workflows werden als JSON definiert (alternativ über die UI oder Terraform). So sieht die Struktur für deine Pipeline aus:^[1]

{
  "name": "weather-to-power-pipeline",
  "schedule": {
    "quartz_cron_expression": "0 0 5,11,17,23 * * ?",
    "timezone_id": "UTC"
  },
  "tasks": [
    {
      "task_key": "ingest_ecmwf",
      "notebook_task": { "notebook_path": "/pipelines/ingest_ecmwf" },
      "job_cluster_key": "ingest_cluster",
      "timeout_seconds": 1800,
      "retry_on_timeout": true,
      "max_retries": 2
    },
    {
      "task_key": "ingest_gfs",
      "notebook_task": { "notebook_path": "/pipelines/ingest_gfs" },
      "job_cluster_key": "ingest_cluster",
      "max_retries": 2
    },
    {
      "task_key": "ingest_icon",
      "notebook_task": { "notebook_path": "/pipelines/ingest_icon" },
      "job_cluster_key": "ingest_cluster",
      "max_retries": 2
    },
    {
      "task_key": "ingest_telemetry",
      "notebook_task": { "notebook_path": "/pipelines/ingest_telemetry" },
      "job_cluster_key": "ingest_cluster",
      "max_retries": 2
    },
    {
      "task_key": "silver_weather",
      "depends_on": [
        {"task_key": "ingest_ecmwf"},
        {"task_key": "ingest_gfs"},
        {"task_key": "ingest_icon"}
      ],
      "notebook_task": { "notebook_path": "/pipelines/bronze_to_silver_weather" },
      "job_cluster_key": "transform_cluster"
    },
    {
      "task_key": "silver_telemetry",
      "depends_on": [{"task_key": "ingest_telemetry"}],
      "notebook_task": { "notebook_path": "/pipelines/bronze_to_silver_telemetry" },
      "job_cluster_key": "transform_cluster"
    },
    {
      "task_key": "compute_features",
      "depends_on": [
        {"task_key": "silver_weather"},
        {"task_key": "silver_telemetry"}
      ],
      "notebook_task": { "notebook_path": "/pipelines/compute_features" },
      "job_cluster_key": "transform_cluster"
    },
    {
      "task_key": "ml_inference",
      "depends_on": [{"task_key": "compute_features"}],
      "notebook_task": { "notebook_path": "/pipelines/ml_inference" },
      "job_cluster_key": "ml_cluster"
    },
    {
      "task_key": "write_gold",
      "depends_on": [{"task_key": "ml_inference"}],
      "notebook_task": { "notebook_path": "/pipelines/write_gold" },
      "job_cluster_key": "transform_cluster"
    }
  ],
  "job_clusters": [
    {
      "job_cluster_key": "ingest_cluster",
      "new_cluster": {
        "num_workers": 0,
        "spark_version": "14.3.x-scala2.12",
        "node_type_id": "Standard_E4s_v3",
        "init_scripts": [{"dbfs": {"destination": "dbfs:/init/install_cfgrib.sh"}}]
      }
    },
    {
      "job_cluster_key": "transform_cluster",
      "new_cluster": {
        "num_workers": 2,
        "spark_version": "14.3.x-scala2.12",
        "node_type_id": "Standard_D4s_v3"
      }
    },
    {
      "job_cluster_key": "ml_cluster",
      "new_cluster": {
        "num_workers": 1,
        "spark_version": "14.3.x-ml-scala2.12",
        "node_type_id": "Standard_E8s_v3"
      }
    }
  ],
  "email_notifications": {
    "on_failure": ["team@company.com"],
    "on_duration_warning_threshold_exceeded": ["team@company.com"]
  },
  "health": {
    "rules": [
      {"metric": "RUN_DURATION_SECONDS", "op": "GREATER_THAN", "value": 3600}
    ]
  }
}

Zentrale Entwurfsentscheidungen

1. Cluster über Tasks hinweg wiederverwenden

Tasks mit demselben job_cluster_key nutzen denselben Cluster — er bleibt zwischen den Tasks am Leben. Das spart Startzeit (ca. 3–5 Min. pro Cluster), bedeutet aber, dass du die Leerlaufzeit zwischen aufeinanderfolgenden Tasks bezahlst.

2. Arten von Abhängigkeiten

// Standard: ALLE Vorgänger müssen erfolgreich sein
"depends_on": [{"task_key": "ingest_ecmwf"}, {"task_key": "ingest_gfs"}]

// Wenn eine Quelle optional ist (z. B. ICON als Nice-to-have):
// "run_if"-Bedingung nutzen oder den optionalen Pfad abtrennen
"run_if": "ALL_SUCCESS"  // Standard
"run_if": "AT_LEAST_ONE_SUCCESS"  // wenn Teildaten akzeptabel sind

3. Retry-Strategien

{
  "max_retries": 2,            // insgesamt 3 Versuche
  "min_retry_interval_millis": 60000,  // 1 Min. Pause zwischen den Versuchen
  "retry_on_timeout": true     // auch bei Timeout erneut versuchen
}

Retries sind bei vorübergehenden Störungen sinnvoll (API-Timeout, verdrängte Spot-Instanz). Sie sind es nicht bei Logikfehlern (fehlerhafter Code) oder bei Datenqualitätsproblemen — die gehören gemeldet, nicht wiederholt.

Strategie für den Zeitplan

Der Zeitplan deiner Pipeline muss zur Verfügbarkeit der Wettermodelle passen:

# Cron: 4× täglich, 5 Stunden nach der Modell-Initialisierung
# "0 0 5,11,17,23 * * ?" = um 05:00, 11:00, 17:00 und 23:00 UTC
"quartz_cron_expression": "0 0 5,11,17,23 * * ?"

Monitoring & Alerting

Eingebautes Databricks-Monitoring

• E-Mail/Webhook bei Fehlern: Sofortige Meldung, wenn ein Task nach allen Retries scheitert
• Health-Rules zur Laufzeit: Meldung, wenn die Pipeline die erwartete Laufzeit überschreitet (Hinweis auf Performance-Regression)
• Lauf-Historie: Erfolgsquote und Laufzeittrends über die Workflows-UI verfolgen

Eigene Prüfungen auf Pipeline-Ebene (letzter Task)

# Letzter Task im Workflow: Ausgabe der Pipeline validieren
from datetime import datetime, timedelta, timezone

def validate_pipeline_output():
    """Assert that Gold table has fresh data after pipeline completes."""
    
    latest_gold = spark.sql("""
        SELECT MAX(produced_at) as latest
        FROM prod.gold.forecast_ts
    """).collect()[0]["latest"]
    
    threshold = datetime.now(timezone.utc) - timedelta(hours=2)
    assert latest_gold >= threshold, \
        f"Gold data stale! Latest: {latest_gold}, expected >= {threshold}"
    
    # Prüfen, ob alle Parks Prognosen haben
    parks_with_forecasts = spark.sql("""
        SELECT COUNT(DISTINCT asset_id)
        FROM prod.gold.forecast_ts
        WHERE produced_at > current_timestamp() - INTERVAL 2 HOURS
          AND asset_type = 'wind_park'
    """).collect()[0][0]
    
    expected_parks = spark.sql("""
        SELECT COUNT(*) FROM prod.silver.dim_assets
        WHERE asset_type = 'wind_park'
    """).collect()[0][0]
    
    assert parks_with_forecasts == expected_parks, \
        f"Missing parks! Got {parks_with_forecasts}/{expected_parks}"
    
    print(f"✓ Pipeline validation passed: {parks_with_forecasts} parks forecasted")

Infrastructure as Code

Konfiguriere Workflows nicht von Hand in der UI. Nutze Infrastructure as Code, damit alles reproduzierbar bleibt:

Prüfe dein Verständnis

Primärquelle

Wie geht es weiter

Die Pipeline ist vollständig orchestriert. Die letzte Lektion behandelt Kosten, Performance & Betrieb — Cluster effizient dimensionieren, Ausgaben steuern, Delta-Tabellen optimieren und Runbooks für den Produktivbetrieb schreiben.

Fragen zu Setups mit mehreren Workflows, zum Auslösen von Pipelines durch externe Events oder zur CI/CD-Integration? Frag deinen Lehrer.

Rechtliches: Impressum