Monitoring, Logging and Cost Management in Azure Databricks

Level: Intermediate / Advanced | Platform: Azure Databricks + Azure Monitor

Table of Contents

1. Why Monitor Databricks?
2. Built-in Monitoring Tools
3. Logging Architecture
4. Configuring Diagnostic Settings
5. Azure Monitor and Log Analytics
6. KQL Queries for Databricks
7. Alerts and Notifications
8. Spark Job Performance Analysis
9. Common Bottlenecks and Solutions
10. Autoscaling and Instance Pools — Optimization
11. Spark Configuration Tuning
12. Cost Management — Levers
13. Spot Instances and Auto-Termination
14. Azure Cost Management — Budgets and Alerts
15. Databricks Usage Dashboard
16. Delta Lake Storage Optimization
17. Automated Reporting
18. Event-Driven Monitoring with Azure Functions
19. Automated Resource Cleanup
20. Dashboards with Azure Synapse / Power BI
21. Summary and Best Practices
22. Glossary

1. Why Monitor Databricks?

1.1 Enterprise Monitoring Challenges

Without adequate monitoring, a Databricks organization is exposed to several critical risks:

mindmap
  root((Risks Without\nMonitoring))
    Performance
      Slow jobs undetected
      Ignored bottlenecks
      SLAs not met
    Reliability
      Undetected failures
      Difficult root cause analysis
      Silently failing pipelines
    Costs
      Idle clusters not stopped
      Over-provisioning
      Budget exceeded without alert
    Compliance
      Suspicious activities untracked
      Impossible audit
      GDPR violations
    Visibility
      Who does what on the workspace
      Which resource consumes what
      What data is accessed

1.2 The Four Dimensions of Databricks Monitoring

2. Built-in Monitoring Tools

2.1 Overview of Available Tools

graph LR
    subgraph "Databricks UI"
        SparkUI[Spark UI\nJobs, Stages, Tasks, DAG]
        Ganglia[Ganglia Metrics\nCPU, RAM, Network]
        EventLog[Event Logs\nCluster Events]
        JobRuns[Job Runs\nRun History]
    end
    
    subgraph "Azure Services"
        Monitor[Azure Monitor\nCentralized Metrics]
        LA[Log Analytics\nKQL Queries]
        CM[Cost Management\nBudgets + Alerts]
        AppIns[Application Insights\nCustom APM]
    end
    
    subgraph "Databricks Account"
        UsageDash[Usage Dashboard\nDBUs per workspace]
        AuditLogs[Audit Logs\nUser activity]
    end
    
    SparkUI & Ganglia & EventLog --> Monitor
    Monitor --> LA
    UsageDash & AuditLogs --> LA

2.2 Spark UI — Navigation and Interpretation

graph LR
    SparkUI[Spark UI] --> Jobs[Jobs Tab\nLists all jobs]
    SparkUI --> Stages[Stages Tab\nExecution stages]
    SparkUI --> Storage[Storage Tab\nCached RDDs/DataFrames]
    SparkUI --> Exec[Executors Tab\nWorker health]
    SparkUI --> SQL[SQL Tab\nSQL/DF queries]
    
    Jobs --> J1[Job ID, Duration\nStatus, Stages]
    Stages --> S1[Input/Output Size\nShuffle Read/Write\nTask Distribution]
    Exec --> E1[Memory/Disk Usage\nGC Time\nTask Metrics]
    SQL --> SQL1[Query Plan\nAQE Details\nPhoton Usage]

2.3 Ganglia Metrics

3. Logging Architecture

3.1 Log Types in Databricks

graph TB
    subgraph "Databricks Logs"
        CL[Cluster Logs\nDriver + Executor stdout/stderr\nInit script logs]
        WL[Workspace Audit Logs\nUser actions, admin changes\nLogin attempts]
        JL[Job Logs\nTask outputs, errors\nPerformance metrics]
        DL[Diagnostic Logs\nPlatform events\nCluster lifecycle]
    end
    
    subgraph "Destinations"
        BLOB["(Azure Blob Storage\nLong-term retention)"]
        ADLS["(ADLS Gen2\nAnalytics-ready)"]
        EH[Event Hub\nReal-time streaming]
        LAW[Log Analytics\nWorkspace]
    end
    
    CL & WL & JL & DL -->|Diagnostic Settings| BLOB & ADLS & EH & LAW

3.2 Available Databricks Log Categories

4. Configuring Diagnostic Settings

4.1 Enable Diagnostic Settings via Azure Portal

# Via Azure CLI — Configure Diagnostic Settings for a Databricks workspace
WORKSPACE_RESOURCE_ID="/subscriptions/{sub-id}/resourceGroups/ETL-RG/providers/Microsoft.Databricks/workspaces/my-workspace"
LAW_RESOURCE_ID="/subscriptions/{sub-id}/resourceGroups/ETL-RG/providers/Microsoft.OperationalInsights/workspaces/databricks-law"

az monitor diagnostic-settings create \
  --name "databricks-diagnostics-complete" \
  --resource "$WORKSPACE_RESOURCE_ID" \
  --workspace "$LAW_RESOURCE_ID" \
  --logs '[
    {"category": "accounts", "enabled": true, "retentionPolicy": {"enabled": true, "days": 90}},
    {"category": "clusters", "enabled": true, "retentionPolicy": {"enabled": true, "days": 90}},
    {"category": "jobs", "enabled": true, "retentionPolicy": {"enabled": true, "days": 90}},
    {"category": "notebook", "enabled": true, "retentionPolicy": {"enabled": true, "days": 90}},
    {"category": "secrets", "enabled": true, "retentionPolicy": {"enabled": true, "days": 365}},
    {"category": "sqlPermissions", "enabled": true, "retentionPolicy": {"enabled": true, "days": 365}},
    {"category": "workspace", "enabled": true, "retentionPolicy": {"enabled": true, "days": 90}}
  ]'

echo "Diagnostic Settings configured successfully"

4.2 Enable Databricks Cluster Logging

# Configure cluster logging via the Databricks API
import requests

workspace_url = "https://adb-xxxx.azuredatabricks.net"
token = dbutils.secrets.get(scope="kv-secrets", key="databricks-token")

# Cluster configuration with logging enabled
cluster_config = {
    "cluster_name": "production-etl-cluster",
    "spark_version": "13.3.x-scala2.12",
    "node_type_id": "Standard_DS4_v2",
    "num_workers": 4,
    "cluster_log_conf": {
        "dbfs": {
            "destination": "dbfs:/cluster-logs/etl-cluster"
        }
    }
}

response = requests.post(
    f"{workspace_url}/api/2.0/clusters/create",
    headers={"Authorization": f"Bearer {token}"},
    json=cluster_config
)

print(f"Cluster created: {response.json().get('cluster_id')}")

4.3 Full Diagnostic Log Categories

# Create a Diagnostic Setting sending all logs to Log Analytics
az monitor diagnostic-settings create \
  --name "databricks-diag" \
  --resource "/subscriptions/<sub-id>/resourceGroups/<rg>/providers/Microsoft.Databricks/workspaces/<ws-name>" \
  --workspace "/subscriptions/<sub-id>/resourceGroups/<rg>/providers/Microsoft.OperationalInsights/workspaces/<law-name>" \
  --logs '[
    {"category": "dbfs",               "enabled": true},
    {"category": "clusters",           "enabled": true},
    {"category": "accounts",           "enabled": true},
    {"category": "jobs",               "enabled": true},
    {"category": "notebook",           "enabled": true},
    {"category": "sql",                "enabled": true},
    {"category": "genie",              "enabled": true},
    {"category": "globalInitScripts",  "enabled": true},
    {"category": "secrets",            "enabled": true},
    {"category": "sqlPermissions",     "enabled": true},
    {"category": "instancePools",      "enabled": true}
  ]'

5. Azure Monitor and Log Analytics

5.1 Azure Monitor Architecture for Databricks

graph TB
    subgraph "Data Sources"
        DB[Azure Databricks\nWorkspace]
        AZ[Azure Infrastructure\nVM, Storage, Network]
        CUSTOM[Custom Metrics\nApplication code]
    end
    
    subgraph "Azure Monitor"
        CM2[Collect\nMetrics + Logs]
        SM[Store\nMetrics DB + Log Analytics]
        ANA[Analyze\nKQL Queries + Workbooks]
        VIZ[Visualize\nDashboards + Power BI]
        ACT[Act\nAlerts + Auto-remediation]
    end
    
    DB -->|Diagnostic Settings| CM2
    AZ --> CM2
    CUSTOM --> CM2
    CM2 --> SM
    SM --> ANA --> VIZ
    SM --> ACT
    ACT --> Alert[Email/Teams/PagerDuty\nWebhook]
    ACT --> Auto[Azure Automation\nAuto-remediation]

5.2 Create a Log Analytics Workspace

# Create the Log Analytics Workspace
az monitor log-analytics workspace create \
  --resource-group "ETL-RG" \
  --workspace-name "databricks-law" \
  --location "eastus2" \
  --sku PerGB2018 \
  --retention-time 90  # 90-day retention

# Get the Workspace ID and Primary Key
LAW_ID=$(az monitor log-analytics workspace show \
  --resource-group "ETL-RG" \
  --workspace-name "databricks-law" \
  --query "customerId" -o tsv)

LAW_KEY=$(az monitor log-analytics workspace get-shared-keys \
  --resource-group "ETL-RG" \
  --workspace-name "databricks-law" \
  --query "primarySharedKey" -o tsv)

echo "Workspace ID: $LAW_ID"
echo "Workspace Key: [REDACTED]"

5.3 Key Metrics to Monitor

6. KQL Queries for Databricks

6.1 Essential KQL Queries

// ══════════════════════════════════════════════════════
// 1. AUDIT: User actions (last 24h)
// ══════════════════════════════════════════════════════
DatabricksAccounts
| where TimeGenerated > ago(24h)
| where ActionName != "aadBased"
| project TimeGenerated, UserName = identity.email, 
          ActionName, RequestParams, Response = response.statusCode
| order by TimeGenerated desc
| take 100

// ══════════════════════════════════════════════════════
// 2. CLUSTERS: Event history
// ══════════════════════════════════════════════════════
DatabricksClusters
| where TimeGenerated > ago(7d)
| project TimeGenerated, ClusterName = requestParams.cluster_name,
          ActionName, RequestedBy = identity.email,
          ClusterId = requestParams.cluster_id
| where ActionName in ("create", "delete", "edit", "start", "terminate")
| order by TimeGenerated desc

// ══════════════════════════════════════════════════════
// 3. JOBS: Job failure rate
// ══════════════════════════════════════════════════════
DatabricksJobs
| where TimeGenerated > ago(7d)
| where ActionName == "runNow" or ActionName == "run"
| extend JobName = requestParams.job_name
| extend Status = response.result_state
| summarize 
    TotalRuns = count(),
    SuccessRuns = countif(Status == "SUCCESS"),
    FailureRuns = countif(Status == "FAILED"),
    CancelledRuns = countif(Status == "CANCELED")
  by JobName
| extend SuccessRate = round(100.0 * SuccessRuns / TotalRuns, 2)
| order by FailureRuns desc

// ══════════════════════════════════════════════════════
// 4. SECURITY: Secret access (security logs)
// ══════════════════════════════════════════════════════
DatabricksSecrets
| where TimeGenerated > ago(24h)
| project TimeGenerated, User = identity.email,
          ActionName, SecretScope = requestParams.scope,
          SecretKey = requestParams.key
| order by TimeGenerated desc

// ══════════════════════════════════════════════════════
// 5. NOTEBOOKS: Notebook activity
// ══════════════════════════════════════════════════════
DatabricksNotebook
| where TimeGenerated > ago(7d)
| summarize 
    Actions = count(),
    UniqueUsers = dcount(identity_email)
  by NotebookPath = requestParams.notebookPath, 
     ActionName
| order by Actions desc
| take 20

// ══════════════════════════════════════════════════════
// 6. PERFORMANCE: Top 10 longest running jobs
// ══════════════════════════════════════════════════════
DatabricksJobs
| where TimeGenerated > ago(30d)
| where ActionName == "runFinished"
| extend DurationMs = todouble(response.execution_duration)
| extend JobName = requestParams.job_name
| summarize 
    AvgDurationMin = round(avg(DurationMs) / 60000, 2),
    MaxDurationMin = round(max(DurationMs) / 60000, 2),
    P95DurationMin = round(percentile(DurationMs, 95) / 60000, 2),
    TotalRuns = count()
  by JobName
| order by AvgDurationMin desc
| take 10

6.2 Advanced KQL Reference Queries

// ── 1. Cluster startup failures (last 24h) ─────────────────────────────────
DatabricksClusters
| where TimeGenerated > ago(24h)
| where ActionName == "create" and isnotempty(TerminationReason)
| summarize FailureCount = count() by ClusterName, TerminationReason
| order by FailureCount desc

// ── 2. Job success/failure rate by hour ────────────────────────────────────
DatabricksJobs
| where TimeGenerated > ago(7d)
| where ActionName in ("runSucceeded", "runFailed")
| summarize
    Successes = countif(ActionName == "runSucceeded"),
    Failures  = countif(ActionName == "runFailed")
    by bin(TimeGenerated, 1h)
| extend FailureRate = round(100.0 * Failures / (Successes + Failures), 2)
| order by TimeGenerated desc

// ── 3. Top 10 users by activity ────────────────────────────────────────────
DatabricksAccounts
| where TimeGenerated > ago(30d)
| summarize ActionCount = count() by User
| top 10 by ActionCount desc

// ── 4. Secret access (security audit) ─────────────────────────────────────
DatabricksSecrets
| where TimeGenerated > ago(7d)
| project TimeGenerated, User, ActionName, ScopeName, KeyName
| order by TimeGenerated desc

// ── 5. SQL query duration by warehouse ────────────────────────────────────
DatabricksSql
| where TimeGenerated > ago(24h)
| where ActionName == "commandSubmit"
| summarize
    AvgDurationMs = avg(DurationMs),
    MaxDurationMs = max(DurationMs),
    P95DurationMs = percentile(DurationMs, 95)
    by WarehouseId
| order by P95DurationMs desc

// ── 6. Cluster autoscaling events ─────────────────────────────────────────
DatabricksClusters
| where TimeGenerated > ago(24h)
| where ActionName in ("resize", "upsize", "downsize")
| project TimeGenerated, ClusterId, ClusterName, ActionName, User
| order by TimeGenerated desc

// ── 7. Alert on repeated job failures ─────────────────────────────────────
DatabricksJobs
| where TimeGenerated > ago(1h)
| where ActionName == "runFailed"
| summarize RecentFailures = count() by JobId
| where RecentFailures >= 3

6.3 Create KQL-Based Alerts

// Alert: Too many job failures in the last hour
// Configure as Scheduled Query Rule in Azure Monitor

DatabricksJobs
| where TimeGenerated > ago(1h)
| where ActionName == "run"
| extend Status = response.result_state
| where Status == "FAILED"
| summarize FailureCount = count() by bin(TimeGenerated, 15m)
| where FailureCount > 5  // Alert if more than 5 failures in 15 min

7. Alerts and Notifications

7.1 Configure Job Notifications in Databricks

# Configure notifications via the Databricks API
import requests

workspace_url = "https://adb-xxxx.azuredatabricks.net"
token = "your_token"

# Job configuration with complete notifications
job_config = {
    "name": "ETL Production Pipeline",
    "tasks": [
        {
            "task_key": "main_etl",
            "notebook_task": {
                "notebook_path": "/Production/ETL/main_pipeline"
            },
            "existing_cluster_id": "cluster-id"
        }
    ],
    "email_notifications": {
        "on_start": [],
        "on_success": ["data-team@company.com"],
        "on_failure": [
            "data-team@company.com",
            "on-call-engineer@company.com"
        ],
        "no_alert_for_skipped_runs": True
    },
    "webhook_notifications": {
        "on_failure": [
            {
                "id": "webhook-id-pagerduty"
            }
        ]
    },
    "notification_settings": {
        "no_alert_for_skipped_runs": True,
        "no_alert_for_canceled_runs": False
    },
    "timeout_seconds": 7200,  # 2-hour max
    "max_retries": 2,
    "min_retry_interval_millis": 300000  # 5 minutes between retries
}

response = requests.post(
    f"{workspace_url}/api/2.0/jobs/create",
    headers={"Authorization": f"Bearer {token}"},
    json=job_config
)

print(f"Job created: {response.json().get('job_id')}")

7.2 Available Alert Types

7.3 Configure a Job Failure Alert

Azure Monitor → Alerts → + Create → Alert rule
  ┌─────────────────────────────────────────────────────┐
  │ SCOPE    : Databricks Workspace                     │
  │ CONDITION: Custom log search                        │
  │   Query  : DatabricksJobs                           │
  │           | where ActionName == "runFailed"         │
  │           | summarize count()                       │
  │   Operator   : Greater than                         │
  │   Threshold  : 0                                    │
  │   Frequency  : 5 minutes                            │
  │ ACTION GROUP : Email + Teams webhook                │
  │ SEVERITY     : 1 (Error)                            │
  │ RULE NAME    : databricks-job-failure-alert         │
  └─────────────────────────────────────────────────────┘

8. Spark Job Performance Analysis

8.1 Performance Analysis Method

flowchart TD
    Start[Slow job detected] --> S1{Identify the slow\nstage in Spark UI}
    S1 --> S2{Data Skew?\nHighly unbalanced tasks}
    S1 --> S3{Too much Shuffle?\nLarge shuffle read/write}
    S1 --> S4{Insufficient memory?\nHigh GC time}
    S1 --> S5{Slow I/O?\nSlow disk reads}
    
    S2 -->|Yes| Fix2[Repartition\nSalt key for skew]
    S3 -->|Yes| Fix3[Broadcast join\nReduce shuffles]
    S4 -->|Yes| Fix4[Increase executor.memory\nOptimize caching]
    S5 -->|Yes| Fix5[Use Delta Lake\nZ-ORDER on filter columns]

8.2 Identifying Bottlenecks in Code

# Enable Spark profiling for analysis
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# Measure execution time of key steps
import time

def timed_operation(name: str, func):
    """Measure the time for a Spark operation."""
    start = time.time()
    result = func()
    duration = time.time() - start
    print(f"⏱️  {name}: {duration:.2f}s")
    return result

# Analyze partition distribution (data skew detection)
def analyze_partition_distribution(df, sample_size=100):
    """Analyze data distribution per partition."""
    from pyspark.sql import functions as F
    
    partition_sizes = df.withColumn(
        "partition_id", F.spark_partition_id()
    ).groupBy("partition_id").count()
    
    stats = partition_sizes.describe("count").toPandas()
    print("Partition distribution:")
    print(stats)
    
    # Detect skew
    max_size = partition_sizes.agg(F.max("count")).collect()[0][0]
    avg_size = partition_sizes.agg(F.avg("count")).collect()[0][0]
    skew_ratio = max_size / avg_size if avg_size > 0 else 0
    
    if skew_ratio > 5:
        print(f"⚠️  DATA SKEW DETECTED! Max/avg ratio: {skew_ratio:.2f}")
        print("   Recommendation: Use repartition() or AQE skew join handling")
    else:
        print(f"✅ Balanced distribution (ratio: {skew_ratio:.2f})")
    
    return skew_ratio

# Analyze the execution plan
def analyze_query_plan(df):
    """Analyze the execution plan to identify expensive operations."""
    print("=== Execution Plan ===")
    df.explain(extended=True)
    
    # Check if a broadcast join is used
    plan_str = df._jdf.queryExecution().simpleString()
    if "BroadcastHashJoin" in plan_str:
        print("✅ Broadcast Join used (optimal for small tables)")
    elif "SortMergeJoin" in plan_str:
        print("⚠️  Sort-Merge Join used (can be slow for large tables)")
        print("   Recommendation: If one table is < 128MB, use broadcast()")

9. Common Bottlenecks and Solutions

9.1 Performance Issue Resolution Guide

9.2 Resolving Data Skew

from pyspark.sql import functions as F

# PROBLEM: Data skew on "category" column (some values very frequent)
# Example: 80% of data has category = "other"

# SOLUTION 1: Salting (add random suffix to key)
def salt_join_fix_skew(df_large, df_small, join_key, n_salt=10):
    """
    Resolves data skew by distributing the join key.
    
    Args:
        df_large: Large DataFrame with skew
        df_small: Small DataFrame (lookup table)
        join_key: Join column with skew
        n_salt: Number of salt partitions
    """
    # Add a random salt value to the large DataFrame
    df_large_salted = df_large.withColumn(
        "salt_key",
        F.concat(F.col(join_key), F.lit("_"), 
                 (F.rand() * n_salt).cast("int").cast("string"))
    )
    
    # Replicate the small DataFrame with all salt values
    salt_values = spark.range(n_salt).select(
        F.col("id").cast("string").alias("salt")
    )
    df_small_replicated = df_small.crossJoin(salt_values).withColumn(
        "salt_key",
        F.concat(F.col(join_key), F.lit("_"), F.col("salt"))
    ).drop("salt")
    
    # Join on salted key
    result = df_large_salted.join(df_small_replicated, "salt_key", "left") \
        .drop("salt_key")
    
    return result

# SOLUTION 2: Use AQE (Adaptive Query Execution) — recommended
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5.0")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")

# SOLUTION 3: Broadcast join for small tables
from pyspark.sql.functions import broadcast

# If the small table is < 128 MB
result = large_df.join(broadcast(small_df), "category")

10. Autoscaling and Instance Pools — Optimization

10.1 Autoscaling — Optimal Configuration

// Cluster configuration with optimized autoscaling
{
  "cluster_name": "production-autoscale-cluster",
  "spark_version": "13.3.x-scala2.12",
  "node_type_id": "Standard_DS4_v2",
  "driver_node_type_id": "Standard_DS4_v2",
  
  "autoscale": {
    "min_workers": 2,
    "max_workers": 20
  },
  
  "autotermination_minutes": 30,
  
  "spark_conf": {
    "spark.databricks.cluster.profile": "serverless",
    "spark.databricks.delta.optimizeWrite.enabled": "true",
    "spark.databricks.enhanced.autoscaling.enabled": "true"
  },
  
  "azure_attributes": {
    "availability": "SPOT_WITH_FALLBACK_AZURE",
    "first_on_demand": 2,
    "spot_bid_max_price": -1
  },
  
  "custom_tags": {
    "team": "data-engineering",
    "environment": "production",
    "cost_center": "CC-1234"
  }
}

10.2 Scalability Strategy Comparison

10.3 Instance Pool — Configuration and Benefits

# Create an Instance Pool via the API
import requests

pool_config = {
    "instance_pool_name": "production-warm-pool",
    "min_idle_instances": 2,          # Always 2 instances available
    "max_capacity": 30,               # Absolute maximum
    "node_type_id": "Standard_DS4_v2",
    "preloaded_spark_versions": [
        "13.3.x-scala2.12"             # Pre-loaded runtime
    ],
    "idle_instance_autotermination_minutes": 60,  # Release after 1h
    "azure_attributes": {
        "availability": "SPOT_WITH_FALLBACK_AZURE",
        "spot_bid_max_price": -1
    },
    "custom_tags": {
        "team": "platform",
        "managed_by": "infra-team"
    }
}

response = requests.post(
    f"{workspace_url}/api/2.0/instance-pools/create",
    headers={"Authorization": f"Bearer {token}"},
    json=pool_config
)

pool_id = response.json().get("instance_pool_id")
print(f"Instance Pool created: {pool_id}")

# Benefit: Cluster startup 1-2 min instead of 5-15 min

11. Spark Configuration Tuning

11.1 Key Configurations for Performance

# Optimized Spark configuration for production
optimal_spark_config = {
    # ── Memory ────────────────────────────────────────────
    "spark.executor.memory": "8g",
    "spark.driver.memory": "4g",
    "spark.executor.memoryOverhead": "2g",    # Off-heap for Python UDFs
    "spark.memory.fraction": "0.8",           # 80% for Spark execution
    "spark.memory.storageFraction": "0.3",    # 30% for cache
    
    # ── CPU / Parallelism ──────────────────────────────────
    "spark.executor.cores": "4",              # 4 cores per executor
    "spark.default.parallelism": "200",       # Default parallelism
    "spark.sql.shuffle.partitions": "200",    # Partitions after shuffle
    
    # ── Adaptive Query Execution (AQE) ────────────────────
    "spark.sql.adaptive.enabled": "true",
    "spark.sql.adaptive.coalescePartitions.enabled": "true",
    "spark.sql.adaptive.coalescePartitions.minPartitionNum": "1",
    "spark.sql.adaptive.skewJoin.enabled": "true",
    
    # ── Delta Lake ────────────────────────────────────────
    "spark.databricks.delta.optimizeWrite.enabled": "true",
    "spark.databricks.delta.autoCompact.enabled": "true",
    "spark.sql.execution.arrow.pyspark.enabled": "true",  # Pandas ↔ Spark
    
    # ── Broadcast ────────────────────────────────────────
    "spark.sql.autoBroadcastJoinThreshold": "134217728",  # 128 MB
    
    # ── I/O ───────────────────────────────────────────────
    "spark.sql.parquet.compression.codec": "snappy",
    "spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version": "2"
}

11.2 Critical Parameters Table

12. Cost Management — Levers

12.1 Main Databricks Cost Levers

graph TB
    subgraph "Databricks Costs"
        C1[DBU Consumption\nDatabricks Units]
        C2[Azure VM Costs\nCompute pricing]
        C3[Storage Costs\nADLS Gen2]
        C4[Network Transfer\nEgress fees]
    end
    
    subgraph "Reduction Levers"
        L1[Spot/Preemptible VMs\n↓60-90% VM cost]
        L2[Auto-Termination\nAvoid idle]
        L3[Job Clusters\nNo idle between runs]
        L4[Serverless\nExact pay-per-use]
        L5[Cluster Policies\nLimit large VMs]
        L6[OPTIMIZE + VACUUM\nReduce storage]
        L7[Lifecycle Policies\nArchive old logs]
    end
    
    C1 --> L3 & L4 & L5
    C2 --> L1 & L2
    C3 --> L6 & L7

12.2 DBU Cost Calculation

DBU (Databricks Units) are the Databricks billing unit:

Impact: A 10-worker All-Purpose cluster (Standard_DS4_v2) active 8h/day costs approximately 10 × 1.75 DBU × $0.55 × 8h = $77/day in DBUs alone, plus Azure VM costs.

13. Spot Instances and Auto-Termination

13.1 Spot Instances (Azure Spot VMs)

Spot Instances use unused Azure capacity at reduced price (-60% to -90%), but can be interrupted:

// Configure Spot Instances in a cluster
{
  "azure_attributes": {
    "availability": "SPOT_WITH_FALLBACK_AZURE",
    "first_on_demand": 1,        // 1 worker On-Demand (never interrupted)
    "spot_bid_max_price": -1     // -1 = market price (most economical)
  }
}

13.2 Auto-Termination — Optimal Configuration

# Configure auto-termination by cluster type
cluster_configs = {
    "dev_cluster": {
        "autotermination_minutes": 15,  # Dev: fast shutdown
        "note": "Short development sessions"
    },
    "analysis_cluster": {
        "autotermination_minutes": 30,  # Analysis: medium session
        "note": "Exploratory analysis sessions"
    },
    "production_allpurpose": {
        "autotermination_minutes": 60,  # Prod: long sessions
        "note": "Shared team cluster"
    },
    "sql_warehouse": {
        "autotermination_minutes": 10,  # SQL: short queries
        "note": "SQL Analytics, ad-hoc queries"
    }
}

for cluster_type, config in cluster_configs.items():
    print(f"{cluster_type}: {config['autotermination_minutes']} min "
          f"({config['note']})")

13.3 Estimated Savings with Auto-Termination

14. Azure Cost Management — Budgets and Alerts

14.1 Create a Budget with Alerts

# Create an Azure budget for Databricks
az consumption budget create \
  --account-name "your-subscription" \
  --budget-name "DatabricksMonthlyBudget" \
  --amount 5000 \
  --category Cost \
  --time-grain Monthly \
  --start-date "2024-01-01" \
  --end-date "2025-12-31" \
  --notification key="alert80pct" \
    enabled=true \
    operator=GreaterThanOrEqualTo \
    threshold=80 \
    contact-emails="finance@company.com" "data-leads@company.com" \
  --notification key="alert100pct" \
    enabled=true \
    operator=GreaterThanOrEqualTo \
    threshold=100 \
    contact-emails="cto@company.com"

14.2 Analyze Costs by Team

# Analyze Databricks costs from Account Console
# Or via Azure Cost Management exports to ADLS Gen2

# Read exported billing data
billing_df = spark.read.parquet(
    "abfss://billing@storageaccount.dfs.core.windows.net/azure-cost-exports/"
)

# Filter on Databricks
databricks_costs = billing_df.filter(
    billing_df["ServiceName"].isin(
        ["Azure Databricks", "Virtual Machines", "Storage"]
    )
)

# Analysis by team (via custom_tags)
from pyspark.sql import functions as F

team_monthly_costs = (
    databricks_costs
    .filter(F.col("Tags.environment") == "production")
    .groupBy(
        F.date_trunc("month", F.col("Date")).alias("Month"),
        F.col("Tags.team").alias("Team"),
        F.col("Tags.cost_center").alias("CostCenter"),
        F.col("ServiceName")
    )
    .agg(
        F.round(F.sum("Cost"), 2).alias("TotalCostUSD"),
        F.count("*").alias("ResourceCount")
    )
    .orderBy("Month", "Team", F.desc("TotalCostUSD"))
)

team_monthly_costs.show(50, truncate=False)

# Identify the most expensive clusters
top_costly_clusters = (
    databricks_costs
    .filter(F.col("ServiceName") == "Azure Databricks")
    .filter(F.col("Tags.cluster_name").isNotNull())
    .groupBy(
        F.col("Tags.cluster_name").alias("ClusterName"),
        F.col("Tags.team").alias("Team")
    )
    .agg(
        F.round(F.sum("Cost"), 2).alias("TotalCostUSD"),
        F.count(F.when(F.col("Date") > F.date_sub(F.current_date(), 7), 1))
          .alias("ActiveLastWeek")
    )
    .orderBy(F.desc("TotalCostUSD"))
    .limit(20)
)

print("Top 20 most expensive clusters:")
top_costly_clusters.show(20, truncate=False)

14.3 Databricks Usage Dashboard via Account Console

The Account Console Usage Dashboard provides:

14.4 Cluster Policies for Cost Control

// Cost-focused cluster policy example
{
  "autotermination_minutes": {
    "type": "fixed",
    "value": 30,
    "hidden": false
  },
  "num_workers": {
    "type": "range",
    "minValue": 1,
    "maxValue": 8
  },
  "node_type_id": {
    "type": "allowlist",
    "values": ["Standard_DS3_v2", "Standard_DS4_v2"],
    "defaultValue": "Standard_DS3_v2"
  },
  "spark_version": {
    "type": "regex",
    "pattern": "^14\\.[0-9]+\\.x-scala2\\.12$"
  },
  "custom_tags.CostCenter": {
    "type": "required"
  },
  "custom_tags.Owner": {
    "type": "required"
  }
}

Policy impacts:

• Auto-termination forced at 30 min → no extended idle clusters
• Max 8 workers → no accidental over-provisioning
• Mandatory tags → guaranteed cost attribution

15. Databricks Usage Dashboard

15.1 Create a Monitoring Dashboard in Databricks

# Automated reporting notebook — run daily
from pyspark.sql import functions as F
from pyspark.sql.window import Window
import datetime

yesterday = (datetime.date.today() - datetime.timedelta(days=1)).strftime("%Y-%m-%d")
last_7_days = (datetime.date.today() - datetime.timedelta(days=7)).strftime("%Y-%m-%d")

print(f"=== Databricks Monitoring Report ===")
print(f"Period: {last_7_days} → {yesterday}\n")

# 1. Job success rate
jobs_stats = spark.sql(f"""
    SELECT 
        job_name,
        COUNT(*) AS total_runs,
        SUM(CASE WHEN status = 'SUCCESS' THEN 1 ELSE 0 END) AS success_runs,
        SUM(CASE WHEN status = 'FAILED' THEN 1 ELSE 0 END) AS failed_runs,
        ROUND(100.0 * SUM(CASE WHEN status = 'SUCCESS' THEN 1 ELSE 0 END) / COUNT(*), 2) AS success_rate,
        AVG(duration_seconds / 60) AS avg_duration_min
    FROM databricks_jobs_history
    WHERE run_date >= '{last_7_days}'
    GROUP BY job_name
    ORDER BY failed_runs DESC
""")

print("1. Job performance (last 7 days):")
jobs_stats.show(20, truncate=False)

# 2. Active clusters and consumption
cluster_usage = spark.sql(f"""
    SELECT 
        cluster_name,
        team,
        SUM(dbus_consumed) AS total_dbus,
        ROUND(SUM(dbus_consumed) * 0.55, 2) AS estimated_cost_usd,
        SUM(uptime_hours) AS total_hours,
        ROUND(SUM(idle_hours) / SUM(uptime_hours) * 100, 1) AS idle_pct
    FROM cluster_usage_daily
    WHERE usage_date >= '{last_7_days}'
    GROUP BY cluster_name, team
    ORDER BY total_dbus DESC
""")

print("\n2. Cluster utilization (last 7 days):")
cluster_usage.show(20, truncate=False)

16. Delta Lake Storage Optimization

16.1 Monitor and Optimize Storage Usage

# Delta Lake storage monitoring script
from delta.tables import DeltaTable
import json

def analyze_delta_table_storage(table_name: str) -> dict:
    """Analyze storage usage for a Delta table."""
    
    # Transaction history
    history = spark.sql(f"DESCRIBE HISTORY {table_name}").collect()
    
    # File sizes
    detail = spark.sql(f"DESCRIBE DETAIL {table_name}").collect()[0]
    num_files = detail["numFiles"]
    size_bytes = detail["sizeInBytes"]
    
    stats = {
        "table_name": table_name,
        "num_files": num_files,
        "size_gb": round(size_bytes / (1024**3), 2),
        "avg_file_size_mb": round(size_bytes / num_files / (1024**2), 2) if num_files > 0 else 0,
        "num_versions": len(history),
        "oldest_version": history[-1]["timestamp"] if history else None,
    }
    
    # Recommendations
    recommendations = []
    
    if stats["avg_file_size_mb"] < 100:
        recommendations.append(
            f"⚠️  Small files ({stats['avg_file_size_mb']} MB) → Run OPTIMIZE"
        )
    
    if stats["num_versions"] > 30:
        recommendations.append(
            f"⚠️  {stats['num_versions']} versions → Run VACUUM RETAIN 168 HOURS"
        )
    
    if stats["size_gb"] > 1:
        recommendations.append(
            "💡 Consider Z-ORDER on frequently filtered columns"
        )
    
    stats["recommendations"] = recommendations
    return stats

# Analyze important tables
tables_to_monitor = [
    "taxicatalog.rides.yellow_taxis",
    "taxicatalog.rides.green_taxis",
    "hr_catalog.workforce_schema.demographics"
]

print("=== Delta Lake Storage Analysis ===\n")
for table in tables_to_monitor:
    try:
        stats = analyze_delta_table_storage(table)
        print(f"Table: {stats['table_name']}")
        print(f"  Size: {stats['size_gb']} GB | Files: {stats['num_files']} "
              f"| Avg size: {stats['avg_file_size_mb']} MB")
        print(f"  Versions retained: {stats['num_versions']}")
        for rec in stats['recommendations']:
            print(f"  {rec}")
        print()
    except Exception as e:
        print(f"❌ Error analyzing {table}: {e}")

# Automate OPTIMIZE and VACUUM on critical tables
def optimize_and_vacuum_table(table_name: str, retention_hours: int = 168):
    """Optimize and clean a Delta table."""
    print(f"Optimizing {table_name}...")
    spark.sql(f"OPTIMIZE {table_name}")
    spark.sql(f"VACUUM {table_name} RETAIN {retention_hours} HOURS")
    print(f"✅ {table_name} optimized and cleaned")

17. Automated Reporting

17.1 Automated Reporting Job

# Weekly cost and performance report job
# Triggered every Monday morning at 8:00am

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from pyspark.sql import functions as F

def generate_weekly_report() -> str:
    """Generate the weekly HTML report."""
    
    last_7_days = (datetime.date.today() - datetime.timedelta(days=7)).isoformat()
    
    # Pipeline success rate
    success_rate = spark.sql(f"""
        SELECT ROUND(100.0 * COUNT(CASE WHEN status = 'SUCCESS' THEN 1 END) / COUNT(*), 1) AS rate
        FROM jobs_history WHERE run_date >= '{last_7_days}'
    """).collect()[0]["rate"]
    
    # Weekly costs
    weekly_cost = spark.sql(f"""
        SELECT ROUND(SUM(estimated_cost_usd), 2) AS cost
        FROM cluster_usage_daily WHERE usage_date >= '{last_7_days}'
    """).collect()[0]["cost"]
    
    # HTML report
    html = f"""
    <html>
    <body>
    <h2>📊 Weekly Azure Databricks Report</h2>
    <p>Week of {last_7_days}</p>
    
    <h3>Key Metrics</h3>
    <table border="1" cellpadding="8">
        <tr><td><b>Job success rate</b></td><td>{success_rate}%</td></tr>
        <tr><td><b>Estimated total cost</b></td><td>${weekly_cost:,.2f} USD</td></tr>
    </table>
    
    <h3>Recommended Actions</h3>
    <ul>
        <li>Run OPTIMIZE on main tables</li>
        <li>Review clusters with idle_pct > 50%</li>
        <li>Review jobs with failure rate > 10%</li>
    </ul>
    </body>
    </html>
    """
    
    return html

# Send the report by email
report_html = generate_weekly_report()
print("Report generated:")
print(report_html[:500])

17.2 Azure Logic Apps Integration

// Azure Logic Apps flow for automated alerts
{
  "definition": {
    "triggers": {
      "Recurrence": {
        "type": "Recurrence",
        "recurrence": {
          "frequency": "Hour",
          "interval": 1
        }
      }
    },
    "actions": {
      "Check_Failed_Jobs": {
        "type": "Http",
        "inputs": {
          "method": "GET",
          "uri": "https://adb-xxxx.azuredatabricks.net/api/2.0/jobs/runs/list",
          "headers": {
            "Authorization": "@concat('Bearer ', parameters('databricks_token'))"
          },
          "queries": {
            "limit": "25",
            "expand_tasks": "true"
          }
        }
      },
      "Condition_Failures": {
        "type": "If",
        "expression": {
          "greater": [
            "@length(body('Check_Failed_Jobs')?['runs'])",
            0
          ]
        },
        "actions": {
          "Send_Alert_Email": {
            "type": "ApiConnection",
            "inputs": {
              "host": {"connection": {"name": "@parameters('$connections')['office365']['connectionId']"}},
              "method": "post",
              "path": "/v2/Mail",
              "body": {
                "To": "oncall@company.com",
                "Subject": "ALERT: Databricks job failures detected",
                "Body": "Databricks jobs have failed. Check immediately."
              }
            }
          }
        }
      }
    }
  }
}

18. Event-Driven Monitoring with Azure Functions

18.1 Azure Function — Automatic Event Response

# Azure Function — Triggered by a Log Analytics event
# Automatically stops clusters that have been idle for more than 2 hours

import logging
import os
import requests
import json
import azure.functions as func
from datetime import datetime, timedelta

def main(event: func.EventGridEvent) -> None:
    """
    Azure Function triggered by an Azure Monitor event.
    Stops Databricks clusters that have been idle for too long.
    """
    logging.info(f"Event received: {event.event_type}")
    
    workspace_url = os.environ["DATABRICKS_HOST"]
    token = os.environ["DATABRICKS_TOKEN"]
    
    headers = {"Authorization": f"Bearer {token}"}
    
    # Get all active clusters
    response = requests.get(
        f"{workspace_url}/api/2.0/clusters/list",
        headers=headers
    )
    
    clusters = response.json().get("clusters", [])
    
    for cluster in clusters:
        cluster_id = cluster.get("cluster_id")
        cluster_name = cluster.get("cluster_name", "")
        state = cluster.get("state")
        
        # Check if the cluster is RUNNING and idle
        if state != "RUNNING":
            continue
        
        # Calculate time since last activity
        last_activity = cluster.get("last_activity_time", 0)
        last_activity_dt = datetime.fromtimestamp(last_activity / 1000)
        idle_duration = datetime.now() - last_activity_dt
        
        # If idle for more than 2 hours and no active jobs
        if idle_duration > timedelta(hours=2):
            num_active = cluster.get("num_active_sessions", 0)
            
            if num_active == 0:
                logging.warning(f"Stopping idle cluster: {cluster_name} "
                                f"(idle for {idle_duration})")
                
                # Stop the cluster
                delete_response = requests.post(
                    f"{workspace_url}/api/2.0/clusters/delete",
                    headers=headers,
                    json={"cluster_id": cluster_id}
                )
                
                if delete_response.status_code == 200:
                    logging.info(f"✅ Cluster {cluster_name} stopped successfully")
                else:
                    logging.error(f"❌ Failed to stop {cluster_name}: {delete_response.text}")

19. Automated Resource Cleanup

19.1 Orphaned Resource Cleanup Script

# Automated cleanup script — run weekly
import requests
from datetime import datetime, timedelta
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

def cleanup_orphaned_resources():
    """Clean up orphaned Databricks resources."""
    
    print("=== Automated Databricks Resource Cleanup ===\n")
    
    cleaned_resources = {
        "terminated_clusters": 0,
        "old_runs_deleted": 0,
        "old_files_deleted": 0
    }
    
    # 1. Delete clusters terminated more than 30 days ago
    print("1. Old terminated clusters...")
    thirty_days_ago = datetime.now() - timedelta(days=30)
    
    clusters = list(w.clusters.list())
    for cluster in clusters:
        if cluster.state and cluster.state.value == "TERMINATED":
            if hasattr(cluster, 'terminated_time') and cluster.terminated_time:
                terminated_dt = datetime.fromtimestamp(cluster.terminated_time / 1000)
                if terminated_dt < thirty_days_ago:
                    try:
                        w.clusters.permanent_delete(cluster_id=cluster.cluster_id)
                        print(f"  🗑️  Cluster deleted: {cluster.cluster_name}")
                        cleaned_resources["terminated_clusters"] += 1
                    except Exception as e:
                        print(f"  ⚠️  Cannot delete {cluster.cluster_name}: {e}")
    
    # 2. Delete old run history (> 90 days)
    print("\n2. Run history...")
    jobs = list(w.jobs.list())
    for job in jobs:
        old_runs = w.jobs.list_runs(
            job_id=job.job_id,
            completed_only=True
        )
        for run in old_runs:
            if run.end_time:
                end_dt = datetime.fromtimestamp(run.end_time / 1000)
                if end_dt < datetime.now() - timedelta(days=90):
                    try:
                        w.jobs.delete_run(run_id=run.run_id)
                        cleaned_resources["old_runs_deleted"] += 1
                    except:
                        pass
    
    print(f"\n=== Cleanup Summary ===")
    for resource, count in cleaned_resources.items():
        print(f"  {resource}: {count} deleted")

cleanup_orphaned_resources()

19.2 Resource Sprawl Prevention Best Practices

20. Dashboards with Azure Synapse / Power BI

20.1 Centralized Monitoring Architecture

graph LR
    subgraph "Sources"
        DB2[Databricks Logs\nDiagnostic Settings]
        AzureLog[Azure Monitor\nMetrics]
        CostData[Azure Cost Management\nBilling]
    end
    
    subgraph "Ingestion"
        ADF2[Azure Data Factory\nAutomated pipeline]
        LA2[Log Analytics\nWorkspace]
    end
    
    subgraph "Storage"
        ADLS2["(ADLS Gen2\nCentralized logs)"]
    end
    
    subgraph "Analytics"
        Synapse[Azure Synapse\nAnalytical SQL]
        PBI[Power BI\nInteractive Dashboards]
    end
    
    DB2 & AzureLog & CostData --> ADF2
    ADF2 --> ADLS2
    ADLS2 --> Synapse
    Synapse --> PBI
    LA2 --> PBI

20.2 SQL Query for Power BI Dashboard

-- SQL view for Databricks monitoring Power BI dashboard
-- To create in Azure Synapse Analytics

CREATE OR REPLACE VIEW databricks_monitoring.vw_daily_kpis AS
SELECT 
    usage_date,
    workspace_name,
    team,
    environment,
    
    -- Jobs
    SUM(total_job_runs) AS total_runs,
    SUM(successful_runs) AS success_runs,
    SUM(failed_runs) AS failed_runs,
    ROUND(100.0 * SUM(successful_runs) / NULLIF(SUM(total_job_runs), 0), 2) AS success_rate,
    
    -- Performance
    AVG(avg_job_duration_min) AS avg_duration_min,
    MAX(max_job_duration_min) AS max_duration_min,
    
    -- Costs
    SUM(dbus_consumed) AS total_dbus,
    ROUND(SUM(estimated_cost_usd), 2) AS total_cost_usd,
    
    -- Efficiency
    ROUND(100.0 * SUM(idle_hours) / NULLIF(SUM(total_hours), 0), 1) AS idle_pct,
    SUM(clusters_auto_terminated) AS auto_terminated

FROM cluster_usage_daily

GROUP BY usage_date, workspace_name, team, environment;

21. Summary and Best Practices

21.1 Monitoring and Cost Checklist

mindmap
  root((Monitoring\nBest Practices))
    Visibility
      Diagnostic Settings enabled
      Log Analytics configured
      Scheduled KQL queries
      Power BI Dashboards
    Alerts
      Job failure notifications
      Budget alerts at 80% and 100%
      Cluster idle > 2h
      OOM errors
    Performance
      AQE enabled
      Regular OPTIMIZE
      Instance Pools for production
      Spot Instances for batch
    Costs
      Auto-termination everywhere
      Job Clusters for prod ETL
      Mandatory tags via policy
      Monthly budget configured
      Weekly review
    Automation
      Weekly cleanup script
      Auto-remediation Functions
      Automated reporting
      Git for notebooks

21.2 Cost Strategy Comparison

22. Glossary

Search Terms

monitoring · logging · cost · management · azure · databricks · spark · data · engineering · analytics · log · monitor · alerts · configuration · dashboard · job · performance · architecture · auto-termination · automated · available · diagnostic · kql · queries