Workflows & Orchestration

This lesson wires everything together: ingestion, Silver, features, ML inference, and Gold — orchestrated as a single production Workflow. By the end, you'll know how to structure task dependencies, handle failures gracefully, configure retries, and set up alerting for your weather-to-power pipeline.

The Complete Pipeline as a Workflow

Workflow Configuration (JSON)

Databricks Workflows are defined as JSON (or via UI/Terraform). Here's the structure for your pipeline:^[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}
    ]
  }
}

Key Design Decisions

1. Cluster reuse across tasks

Tasks that share a job_cluster_key reuse the same cluster — it stays alive between tasks. This saves startup time (~3–5 min per cluster) but means you pay for idle time between sequential tasks.

2. Dependency modes

// Default: ALL upstream must succeed
"depends_on": [{"task_key": "ingest_ecmwf"}, {"task_key": "ingest_gfs"}]

// If one source is optional (e.g., ICON is nice-to-have):
// Use "run_if" condition or separate the optional path
"run_if": "ALL_SUCCESS"  // default
"run_if": "AT_LEAST_ONE_SUCCESS"  // if partial data is acceptable

3. Retry policies

{
  "max_retries": 2,            // total 3 attempts
  "min_retry_interval_millis": 60000,  // wait 1 min between retries
  "retry_on_timeout": true     // also retry if task hits timeout
}

Retries make sense for transient failures (API timeout, spot instance preemption). They don't make sense for logic errors (bad code) or data quality failures (should alert, not retry).

Scheduling Strategy

Your pipeline schedule must align with weather model availability:

# Cron: 4× daily, 5 hours after model init
# "0 0 5,11,17,23 * * ?" = at 05:00, 11:00, 17:00, 23:00 UTC
"quartz_cron_expression": "0 0 5,11,17,23 * * ?"

Monitoring & Alerting

Built-in Databricks monitoring

• Email/webhook on failure: Immediate alert when any task fails after retries exhausted
• Duration health rules: Alert if pipeline exceeds expected runtime (indicates performance regression)
• Run history: Track success rate, duration trends via the Workflows UI

Custom pipeline-level checks (final task)

# Final task in workflow: validate pipeline output
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}"
    
    # Check all parks have forecasts
    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

Don't configure Workflows manually in the UI. Use infrastructure-as-code for reproducibility:

Check Your Understanding

Primary Source

What's Next

The pipeline is fully orchestrated. The final lesson covers Cost, Performance & Operations — sizing clusters efficiently, controlling spend, optimising Delta tables, and operational runbooks for your production system.

Questions about multi-workflow setups, triggering pipelines from external events, or CI/CD integration? Ask your teacher.

Rechtliches: Impressum