databricks-observability

name	databricks-observability
description	Set up comprehensive observability for Databricks with metrics, traces, and alerts. Use when implementing monitoring for Databricks jobs, setting up dashboards, or configuring alerting for pipeline health. Trigger with phrases like "databricks monitoring", "databricks metrics", "databricks observability", "monitor databricks", "databricks alerts", "databricks logging".
allowed-tools	Read, Write, Edit, Bash(databricks:*)
version	1.0.0
license	MIT
author	Jeremy Longshore <[e-mail]>

Databricks Observability

Overview

Set up comprehensive observability for Databricks workloads.

Prerequisites

• Access to system tables
• SQL Warehouse for dashboards
• Notification destinations configured
• Alert recipients defined

Metrics Collection

Key Metrics

Metric	Source	Description
Job success rate	system.lakeflow.job_run_timeline	% of successful job runs
Job duration	system.lakeflow.job_run_timeline	Run time in minutes
Cluster utilization	system.compute.cluster_events	CPU/memory usage
Data freshness	table history	Hours since last update
DBU consumption	system.billing.usage	Cost tracking

Instructions

Step 1: System Tables Access

-- Enable system tables (workspace admin)
-- Check available system tables
SELECT * FROM system.information_schema.tables
WHERE table_schema = 'billing' OR table_schema = 'lakeflow';

-- Job run history
SELECT
    job_id,
    job_name,
    run_id,
    start_time,
    end_time,
    result_state,
    error_message,
    (end_time - start_time) / 60000 as duration_minutes
FROM system.lakeflow.job_run_timeline
WHERE start_time > current_timestamp() - INTERVAL 24 HOURS
ORDER BY start_time DESC;

-- Cluster events
SELECT
    cluster_id,
    timestamp,
    type,
    details
FROM system.compute.cluster_events
WHERE timestamp > current_timestamp() - INTERVAL 24 HOURS
ORDER BY timestamp DESC;

Step 2: Create Monitoring Views

-- Job health summary view
CREATE OR REPLACE VIEW monitoring.job_health_summary AS
SELECT
    job_name,
    COUNT(*) as total_runs,
    SUM(CASE WHEN result_state = 'SUCCESS' THEN 1 ELSE 0 END) as successes,
    SUM(CASE WHEN result_state = 'FAILED' THEN 1 ELSE 0 END) as failures,
    ROUND(SUM(CASE WHEN result_state = 'SUCCESS' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) as success_rate,
    AVG((end_time - start_time) / 60000) as avg_duration_minutes,
    PERCENTILE((end_time - start_time) / , )  p95_duration_minutes,
    (start_time)  last_run_time,
    (  result_state    start_time )  last_failure_time
 system.lakeflow.job_run_timeline
 start_time  ()    DAYS
  job_name;


  REPLACE  monitoring.data_freshness 

    table_catalog,
    table_schema,
    table_name,
    (commit_timestamp)  last_update,
    TIMESTAMPDIFF(, (commit_timestamp), ())  hours_since_update,
    
         TIMESTAMPDIFF(, (commit_timestamp), ())    
         TIMESTAMPDIFF(, (commit_timestamp), ())    
         TIMESTAMPDIFF(, (commit_timestamp), ())    
         
      freshness_status
 system.information_schema.table_history
  table_catalog, table_schema, table_name;


  REPLACE  monitoring.daily_costs 

    (usage_date)  ,
    workspace_id,
    sku_name,
    usage_type,
    (usage_quantity)  total_dbus,
    (usage_quantity  list_price)  estimated_cost
 system.billing.usage
 usage_date  ()    DAYS
  (usage_date), workspace_id, sku_name, usage_type
   , estimated_cost ;

Step 3: Configure Alerts

-- Alert: Job failure
CREATE ALERT job_failure_alert
AS SELECT
    job_name,
    run_id,
    error_message,
    start_time
FROM system.lakeflow.job_run_timeline
WHERE result_state = 'FAILED'
  AND start_time > current_timestamp() - INTERVAL 15 MINUTES
SCHEDULE CRON '*/15 * * * *'
NOTIFICATIONS (
    email_addresses = ['[e-mail]'],
    webhook_destinations = ['slack-alerts']
);

-- Alert: Long-running jobs
CREATE ALERT long_running_job_alert
AS SELECT
    job_name,
    run_id,
    start_time,
    TIMESTAMPDIFF(MINUTE, start_time, current_timestamp()) as running_minutes
FROM system.lakeflow.job_run_timeline
WHERE end_time IS NULL
  AND TIMESTAMPDIFF(MINUTE, start_time, current_timestamp()) > 120
SCHEDULE CRON '*/30 * * * *'
NOTIFICATIONS (
    email_addresses = ['[e-mail]']
);

-- Alert: Data freshness SLA breach
CREATE ALERT data_freshness_sla
AS SELECT
    table_name,
    hours_since_update
FROM monitoring.data_freshness
WHERE table_schema = 'gold'
   hours_since_update  
SCHEDULE CRON 
NOTIFICATIONS (
    email_addresses  []
);


 ALERT daily_cost_spike
 
    ,
    estimated_cost,
    (estimated_cost)  (  )  prev_day_cost,
    (estimated_cost  (estimated_cost)  (  )) 
        ((estimated_cost)  (  ), )    percent_change
 monitoring.daily_costs
   ()  
 percent_change    
SCHEDULE CRON 
NOTIFICATIONS (
    email_addresses  []
);

Step 4: Structured Logging

# src/utils/logging.py
import logging
import json
from datetime import datetime
from typing import Any

class StructuredLogger:
    """Structured logging for Databricks notebooks."""

    def __init__(self, job_name: str, run_id: str = None):
        self.job_name = job_name
        self.run_id = run_id or str(datetime.now().timestamp())
        self.logger = logging.getLogger(job_name)
        self.logger.setLevel(logging.INFO)

        # JSON formatter
        handler = logging.StreamHandler()
        handler.setFormatter(JsonFormatter())
        self.logger.addHandler(handler)

    def _log(self, level: str, message: str, **context):
        """Log with structured context."""
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "job_name": self.job_name,
            "run_id": self.run_id,
            "level": level,
            "message": message,
            **context
        }
        getattr(self.logger, level.lower())(json.dumps(log_entry))

    def info(self, message: , **context):
        ._log(, message, **context)

     ():
        ._log(, message, **context)

     ():
        
        ._log(, , metric_name=name, metric_value=value, **tags)

 (logging.Formatter):
    
     ():
         record.getMessage()


logger = StructuredLogger(, dbutils.notebook.entry_point.getDbutils().notebook().getContext().runId())
logger.info(, source=)
logger.metric(, , table=)

Step 5: Custom Metrics Dashboard

# src/monitoring/dashboard.py
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.sql import (
    Dashboard,
    Widget,
    Query,
)

def create_monitoring_dashboard(w: WorkspaceClient) -> str:
    """Create operational monitoring dashboard."""

    # Create dashboard
    dashboard = w.dashboards.create(
        name="Data Platform Monitoring",
        tags=["monitoring", "operations"],
    )

    # Job Success Rate Widget
    job_success_query = """
    SELECT
        DATE(start_time) as date,
        job_name,
        ROUND(SUM(CASE WHEN result_state = 'SUCCESS' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) as success_rate
    FROM system.lakeflow.job_run_timeline
    WHERE start_time > current_timestamp() - INTERVAL 7 DAYS
    GROUP BY DATE(start_time), job_name
    ORDER BY date, job_name
    """

    # Add widgets...
    # (Dashboard API implementation)

    return dashboard.id

# Generate Grafana dashboard JSON
def generate_grafana_dashboard() -> dict:
    """Generate Grafana dashboard configuration."""
    return {
        "dashboard": {
            "title": "Databricks Monitoring",
            "panels": [
                {
                    "title": "Job Success Rate",
                    "type": "timeseries",
                    "targets": [{
                        "rawSql": """
                            SELECT
                                start_time as time,
                                success_rate
                            FROM monitoring.job_health_summary
                        """
                    }]
                },
                {
                    : ,
                    : ,
                    : [{
                        : 
                    }]
                }
            ]
        }
    }

Step 6: Integration with External Monitoring

# src/monitoring/external.py
import requests
from dataclasses import dataclass

@dataclass
class MetricPoint:
    name: str
    value: float
    tags: dict
    timestamp: int = None

class DatadogExporter:
    """Export metrics to Datadog."""

    def __init__(self, api_key: str, app_key: str):
        self.api_key = api_key
        self.app_key = app_key
        self.base_url = "https://api.datadoghq.com/api/v2"

    def send_metrics(self, metrics: list[MetricPoint]):
        """Send metrics to Datadog."""
        series = []
        for m in metrics:
            series.append({
                "metric": f"databricks.{m.name}",
                "points": [[m.timestamp or int(time.time()), m.value]],
                "tags": [f"{k}:{v}" for k, v in m.tags.items()]
            })

        response = requests.post(
            f"{self.base_url}/series",
            headers={
                "DD-API-KEY": .api_key,
                : .app_key,
            },
            json={: series}
        )
         response.status_code == 


exporter = DatadogExporter(api_key, app_key)
exporter.send_metrics([
    MetricPoint(, , {: , : }),
    MetricPoint(, , {: , : }),
])

Output

• System table queries configured
• Monitoring views created
• SQL alerts active
• Structured logging implemented
• External integrations ready

Error Handling

Issue	Cause	Solution
System tables unavailable	Feature not enabled	Contact admin to enable
Alert not triggering	Wrong schedule	Check cron expression
Missing metrics	Query timeout	Optimize query or increase warehouse
High cardinality	Too many tags	Reduce label dimensions

Examples

Quick Health Check Query

SELECT
    job_name,
    success_rate,
    avg_duration_minutes,
    last_run_time
FROM monitoring.job_health_summary
WHERE success_rate < 95
ORDER BY success_rate ASC;

Resources

• Databricks System Tables
• SQL Alerts
• Dashboards

Next Steps

For incident response, see databricks-incident-runbook.