Monitoring und Alerting für Databricks-Workloads: Kompletter Guide

# Create a job with email alerts via CLI
databricks jobs create --json '{
  "name": "daily_etl",
  "email_notifications": {
    "on_failure": ["oncall@company.com"],
    "on_success": [],
    "on_start": [],
    "no_alert_for_skipped_runs": true
  },
  "webhook_notifications": {
    "on_failure": [{"id": "your-webhook-id"}]
  }
}'

# Set duration alert via Python SDK
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

w.jobs.update(
    job_id=12345,
    new_settings={
        "health": {
            "rules": [
                {
                    "metric": "RUN_DURATION_SECONDS",
                    "op": "GREATER_THAN",
                    "value": 3600  # Alert if job runs longer than 1 hour
                }
            ]
        }
    }
)

# Create a Webhook destination (Databricks Notification Destinations API)
import requests

headers = {
    "Authorization": f"Bearer {DATABRICKS_TOKEN}",
    "Content-Type": "application/json"
}

# Create Slack webhook destination
payload = {
    "display_name": "Data Engineering Slack",
    "slack": {
        "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"
    }
}

response = requests.post(
    f"{DATABRICKS_HOST}/api/2.0/notification-destinations",
    headers=headers,
    json=payload
)
destination_id = response.json()["id"]
print(f"Webhook destination created: {destination_id}")

# init_script.sh — install and configure collectd to ship to CloudWatch
# Add this as a cluster init script in Databricks

cat > /etc/collectd/collectd.conf << EOF
LoadPlugin ganglia
LoadPlugin write_cloudwatch

<Plugin ganglia>
  Host "localhost"
  Port "8649"
</Plugin>

<Plugin write_cloudwatch>
  Region "us-east-1"
  Namespace "Databricks/Clusters"
</Plugin>
EOF

service collectd restart

# Set auto-termination via CLI
databricks clusters edit --json '{
  "cluster_id": "0123-456789-abc123",
  "autotermination_minutes": 30
}'

# spark_metrics_config.py — add to your cluster Spark config
spark_conf = {
    "spark.metrics.conf.*.sink.prometheussink.class":
        "org.apache.spark.metrics.sink.PrometheusServlet",
    "spark.metrics.conf.*.sink.prometheussink.path": "/metrics/prometheus",
    "spark.ui.prometheus.enabled": "true"
}

from pyspark import SparkContext
from pyspark.listener import SparkListener

class JobMetricsListener(SparkListener):
    def __init__(self):
        self.job_start_times = {}
        self.job_durations = {}

    def onJobStart(self, job_start):
        job_id = job_start.jobId()
        self.job_start_times[job_id] = job_start.time()
        print(f"Job {job_id} started at {job_start.time()}")

    def onJobEnd(self, job_end):
        job_id = job_end.jobId()
        duration_ms = job_end.time() - self.job_start_times.get(job_id, job_end.time())
        self.job_durations[job_id] = duration_ms
        status = "succeeded" if job_end.jobResult().toString() == "JobSucceeded" else "failed"
        print(f"Job {job_id} {status} in {duration_ms}ms")
        # Ship to your metrics backend here

listener = JobMetricsListener()
sc = spark.sparkContext
sc._jsc.sc().addSparkListener(listener._jlistener)

# datadog_init_script.sh — Databricks cluster init script
DD_AGENT_MAJOR_VERSION=7 DD_API_KEY=$DD_API_KEY DD_SITE="datadoghq.com" \
  bash -c "$(curl -L https://s3.amazonaws.com/dd-agent/scripts/install_script_agent7.sh)"

# Configure Spark integration
cat > /etc/datadog-agent/conf.d/spark.d/conf.yaml << EOF
init_config:

instances:
  - spark_url: http://localhost:4040
    spark_cluster_mode: spark_standalone_mode
    cluster_name: databricks_cluster
    streaming_metrics: true
    histogram_metrics: true
EOF

service datadog-agent restart

# data_quality_checks.py
from pyspark.sql import functions as F

def check_row_count(table_name: str, min_rows: int, date_col: str = "date",
                    date_val: str = None):
    df = spark.table(table_name)
    if date_val:
        df = df.filter(F.col(date_col) == date_val)

    count = df.count()
    if count < min_rows:
        raise ValueError(
            f"Data quality FAIL: {table_name} has {count} rows "
            f"(expected >= {min_rows}) for {date_val}"
        )
    print(f"  OK {table_name}: {count} rows")

# Run as part of your pipeline
check_row_count("catalog.schema.events", min_rows=100_000, date_val="2024-01-15")
check_row_count("catalog.schema.users",  min_rows=1_000,   date_val="2024-01-15")

import json
from datetime import date

def detect_schema_drift(table_name: str, expected_schema_path: str):
    current_schema = spark.table(table_name).schema.jsonValue()

    with open(expected_schema_path) as f:
        expected_schema = json.load(f)

    current_fields = {f["name"]: f["type"] for f in current_schema["fields"]}
    expected_fields = {f["name"]: f["type"] for f in expected_schema["fields"]}

    added = set(current_fields) - set(expected_fields)
    removed = set(expected_fields) - set(current_fields)

    if added or removed:
        raise ValueError(
            f"Schema drift detected in {table_name}:\n"
            f"  Added: {added}\n"
            f"  Removed: {removed}"
        )
    print(f"  Schema OK: {table_name}")

Severity: P2
Runbook:
1. Databricks-Workflow-Run-Details prüfen (UI -> Workflows -> Run History)
2. Driver-Logs nach Stack-Trace inspizieren
3. Prüfen, ob Upstream-Tabellen aktualisiert wurden (DESCRIBE HISTORY)
4. Cluster-Metriken zum Failure-Zeitpunkt prüfen (Ganglia -> Cluster Events)
5. Re-Run mit --repair-run, wenn Daten intakt

Severity: P3
Runbook:
1. Spark-UI öffnen -> Stages -> langsamste Stage identifizieren
2. Shuffle Read/Write Bytes prüfen (exzessiv = Data Skew oder fehlendes Partition Pruning)
3. Prüfen, ob Input-Datenvolumen gewachsen ist (Row Count vs Historie)
4. Recent Code Changes reviewen (git log)
5. Cluster-Preemption prüfen (Cluster Events -> Spot-Interruptions)