Real-Time Data Processing with Azure Databricks

Platform: Azure Databricks + Azure Event Hub

Table of Contents

1. Batch vs Real-Time — Fundamentals
2. Azure Event Hub — Architecture and Components
3. Event Hub vs Apache Kafka
4. Real-Time Streaming Use Cases
5. Environment Setup
6. Spark Structured Streaming — Introduction
7. Connecting Databricks to Azure Event Hub
8. Simulated IoT Data Producer
9. Spark Structured Streaming Consumer
10. Schema Evolution and Late-Arriving Data
11. Checkpointing and Fault Tolerance
12. Temporal Aggregations — Windowed Operations
13. Stateful Processing — Complex Alerts
14. Real-Time Anomaly Detection (ML)
15. Data Stream Enrichment
16. Autoscaling for Streaming Workloads
17. Monitoring Streaming Jobs
18. Troubleshooting Latency Issues
19. Streaming Cost Optimization
20. Real-Time Reference Architecture
21. Summary and Best Practices
22. Glossary

Module 1. Batch vs Real-Time — Fundamentals

1.1 Core Differences

timeline
    title Batch vs Real-Time Processing Cycle
    
    "00:00" : Batch Job started\n(data collected throughout the day)
    "02:00" : Batch Job completed\nReports available (delay: 2-14h)
    
    "Real-Time" : Event received
               : Processed in < 1 second
               : Result available immediately

1.2 Detailed Comparison Table

1.3 When to Choose Streaming?

flowchart TD
    Start[New use case] --> Q1{Does latency\nconsume value?}
    Q1 -->|No: delay acceptable| Batch[Batch Processing\nSimpler and more economical]
    Q1 -->|Yes: immediacy required| Q2{Required latency}
    Q2 -->|Minutes to hours| Micro[Micro-batch\nSpark Structured Streaming\n30s - 5min]
    Q2 -->|Seconds| NRT[Near Real-Time\nSpark Streaming\n1-30s]
    Q2 -->|Milliseconds| RT[True Real-Time\nKafka Streams, Flink]
    
    Micro --> EH[Azure Event Hub\n+ Databricks Structured Streaming]
    NRT --> EH

Module 2. Azure Event Hub — Architecture and Components

2.1 Event Hub Architecture

graph LR
    subgraph "Producers"
        IOT[IoT Sensors\nAzure IoT Hub]
        APP[Applications\nMicroservices]
        SVC[Web Services\nAPI Events]
        LOGS[System Logs\nDiagnostics]
    end
    
    subgraph "Azure Event Hub Namespace"
        subgraph "Event Hub: device-telemetry"
            P0[Partition 0\nOrdered messages]
            P1[Partition 1\nOrdered messages]
            P2[Partition 2\nOrdered messages]
            P3[Partition N\nOrdered messages]
        end
        
        CG1[Consumer Group:\nstream-analytics]
        CG2[Consumer Group:\ndatabricks-streaming]
        CG3[Consumer Group:\narchiving]
    end
    
    subgraph "Consumers"
        DB[Azure Databricks\nStructured Streaming]
        SA[Stream Analytics\nSQL Aggregations]
        ADLS[ADLS Gen2\nAutomatic Capture]
    end
    
    IOT & APP & SVC & LOGS --> P0 & P1 & P2 & P3
    P0 & P1 & P2 & P3 --> CG1 & CG2 & CG3
    CG1 --> SA
    CG2 --> DB
    CG3 --> ADLS

2.2 Event Hub Components

2.3 Create an Event Hub with Azure CLI

# Variables
RESOURCE_GROUP="telemetry-rg"
LOCATION="westus2"
NAMESPACE="device-telemetry-ns"
EVENT_HUB_NAME="device-telemetry"

# Create the namespace
az eventhubs namespace create \
  --name "$NAMESPACE" \
  --resource-group "$RESOURCE_GROUP" \
  --location "$LOCATION" \
  --sku Standard \
  --enable-auto-inflate true \
  --maximum-throughput-units 10

# Create the Event Hub
az eventhubs eventhub create \
  --name "$EVENT_HUB_NAME" \
  --namespace-name "$NAMESPACE" \
  --resource-group "$RESOURCE_GROUP" \
  --partition-count 4 \
  --message-retention 7

# Create a Consumer Group for Databricks
az eventhubs eventhub consumer-group create \
  --eventhub-name "$EVENT_HUB_NAME" \
  --namespace-name "$NAMESPACE" \
  --resource-group "$RESOURCE_GROUP" \
  --name "databricks-reader-group"

# Retrieve the connection string
az eventhubs namespace authorization-rule keys list \
  --resource-group "$RESOURCE_GROUP" \
  --namespace-name "$NAMESPACE" \
  --name RootManageSharedAccessKey \
  --query primaryConnectionString -o tsv

Module 3. Event Hub vs Apache Kafka

3.1 Full Comparison

Tip: Azure Event Hub is 100% compatible with the Kafka API! Existing Kafka applications can connect to Event Hub without any code changes — simply replace the bootstrap.servers.

3.2 Event Hub with Kafka Protocol

# Connecting a Kafka application to Event Hub (without changing the code!)
kafka_config = {
    "kafka.bootstrap.servers": 
        "device-telemetry-ns.servicebus.windows.net:9093",
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.mechanism": "PLAIN",
    "kafka.sasl.jaas.config": 
        f"org.apache.kafka.common.security.plain.PlainLoginModule required "
        f"username='$ConnectionString' "
        f"password='{event_hub_connection_string}';",
    "subscribe": "device-telemetry",  # Event Hub name = Kafka topic
}

# Read using the Kafka connector
df_kafka = spark.readStream \
    .format("kafka") \
    .options(**kafka_config) \
    .load()

Module 4. Real-Time Streaming Use Cases

4.1 Industries and Use Cases

mindmap
  root((Real-Time\nStreaming))
    IoT & Industry
      Machine monitoring
      Temperature/pressure alerts
      Predictive maintenance
      Production line quality control
    Finance
      Fraud detection
      High-frequency trading
      Risk monitoring
      Real-time compliance
    E-commerce & Retail
      Live recommendations
      Real-time inventory
      User behavior analysis
      Dynamic promotions
    Healthcare
      Patient monitoring
      Vital sign alerts
      Emergency triage
      Pharmaceutical supply chain
    Transportation
      GPS tracking
      Route optimization
      Incident reporting
      Automated tolling
    Social Media
      Trending topics
      Sentiment analysis
      Content moderation
      Engagement analytics

Module 5. Environment Setup

5.1 Storing Event Hub Secrets with Databricks CLI

# Install the Databricks CLI
pip install databricks-cli

# Authenticate
databricks configure --token
# Enter: Host: https://adb-xxxx.azuredatabricks.net
# Enter: Token: dapi_xxxxx

# Create a secret scope (pointing to Azure Key Vault)
databricks secrets create-scope \
  --scope "eventhub-secrets" \
  --scope-backend-type AZURE_KEYVAULT \
  --resource-id "/subscriptions/{sub-id}/resourceGroups/RG/providers/Microsoft.KeyVault/vaults/my-kv" \
  --dns-name "https://my-kv.vault.azure.net/"

# OR create a native Databricks scope
databricks secrets create-scope --scope "eventhub-secrets"

# Add the connection secret
databricks secrets put \
  --scope "eventhub-secrets" \
  --key "eventhub-conn-str" \
  --string-value "Endpoint=sb://device-telemetry-ns.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=xxxxx"

# Verify
databricks secrets list --scope "eventhub-secrets"

5.2 Configuration in Databricks

# In a Databricks notebook
# Retrieve the Event Hub connection string from secrets
event_hub_conn_str = dbutils.secrets.get(
    scope="eventhub-secrets",
    key="eventhub-conn-str"
)

# Event Hub configuration for Spark
event_hub_name = "device-telemetry"

# Expected format by the Event Hub connector for Databricks
eventhubs_conf = {
    "eventhubs.connectionString": 
        spark._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(
            event_hub_conn_str
        ),
    "eventhubs.eventHubName": event_hub_name,
    "eventhubs.consumerGroup": "databricks-reader-group",
    "eventhubs.startingPosition": 
        '{"offset": "-1", "seqNo": -1, "enqueuedTime": null, "isInclusive": true}'
}

print("Event Hub configuration ready!")
print(f"Event Hub: {event_hub_name}")

Module 6. Spark Structured Streaming — Introduction

6.1 Mental Model: Unbounded Table

graph LR
    subgraph "Batch Model"
        BD["(Bounded table\n100,000 rows\nfixed)"]
        BQ[SELECT + GROUP BY]
        BR[Static\nresult]
        BD --> BQ --> BR
    end
    
    subgraph "Streaming Model"
        ST[Unbounded table\nGrows infinitely\nnew events]
        SQ[SAME\nSELECT + GROUP BY]
        SO[Result\ncontinuously updated]
        ST -->|New rows| SQ
        SQ -->|Micro-batch trigger| SO
    end
    
    note[Same DataFrame API\nfor batch AND streaming!]

6.2 Output Modes

6.3 Trigger Types

Module 7. Connecting Databricks to Azure Event Hub

7.1 Installing the Maven Connector

# The Event Hub connector must be installed on the cluster
# In cluster configuration -> Libraries -> Maven:
# Coordinates: com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22
# (or a version compatible with your Spark version)

# Alternative via notebooks.install_package (if available)
# %pip install azure-eventhub  # For the Python producer

7.2 Reading an Event Hub Stream

from pyspark.sql import functions as F
from pyspark.sql.types import *

# Retrieve credentials
event_hub_conn_str = dbutils.secrets.get(
    scope="eventhub-secrets",
    key="eventhub-conn-str"
)

# Complete Event Hub stream configuration
eh_conf = {
    "eventhubs.connectionString": 
        sc._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(
            event_hub_conn_str
        ),
    "eventhubs.eventHubName": "device-telemetry",
    "eventhubs.consumerGroup": "databricks-reader-group",
    # Read from the beginning (use with caution in production)
    "eventhubs.startingPosition": 
        '{"offset": "-1", "seqNo": -1, "enqueuedTime": null, "isInclusive": true}',
    # Or read from now (recommended for new jobs)
    # "eventhubs.startingPosition": '{"offset": "@latest"}'
}

# Create the stream
raw_stream_df = spark.readStream \
    .format("eventhubs") \
    .options(**eh_conf) \
    .load()

# Raw Event Hub DataFrame structure
print("Event Hub columns:")
raw_stream_df.printSchema()
# root
#  |-- body: binary (message data, base64 encoded)
#  |-- partition: string
#  |-- offset: string
#  |-- sequenceNumber: long
#  |-- enqueuedTime: timestamp
#  |-- publisher: string
#  |-- partitionKey: string
#  |-- properties: map<string,json>
#  |-- systemProperties: map<string,json>

Module 8. Simulated IoT Data Producer

8.1 Python Producer — IoT Sensor Simulation

# IoT producer script — run in a Databricks notebook
# Simulates temperature and humidity sensors

from azure.eventhub import EventHubProducerClient, EventData
import json
import time
import random
from datetime import datetime, timedelta

def create_iot_producer():
    """Create and return an IoT Event Hub producer."""
    
    conn_str = dbutils.secrets.get(
        scope="eventhub-secrets",
        key="eventhub-conn-str"
    )
    
    return EventHubProducerClient.from_connection_string(
        conn_str=conn_str,
        eventhub_name="device-telemetry"
    )

def send_device_readings(
    num_devices: int = 5,
    readings_per_second: int = 1,
    duration_seconds: int = 60,
    add_late_events: bool = False,
    anomaly_probability: float = 0.05
):
    """
    Send IoT device readings to Event Hub.
    
    Args:
        num_devices: Number of simulated devices
        readings_per_second: Transmission frequency
        duration_seconds: Simulation duration
        add_late_events: Simulate late-arriving events (20% with -5min offset)
        anomaly_probability: Probability of an anomaly (temperature > 85 C)
    """
    
    producer = create_iot_producer()
    
    print(f"=== Starting IoT simulation ===")
    print(f"Devices: {num_devices}")
    print(f"Duration: {duration_seconds}s")
    print(f"Frequency: {readings_per_second}/s")
    
    messages_sent = 0
    start_time = time.time()
    
    with producer:
        while time.time() - start_time < duration_seconds:
            
            # Create a batch of messages
            event_data_batch = producer.create_batch()
            
            for device_num in range(1, num_devices + 1):
                device_id = f"device-{device_num:03d}"
                
                # Simulate a random anomaly
                is_anomaly = random.random() < anomaly_probability
                
                # Generate values
                base_temp = 65.0
                temperature = (
                    random.uniform(80, 95) if is_anomaly 
                    else random.uniform(55, 75)
                )
                humidity = random.uniform(30, 80)
                
                # Timestamp (with simulated latency if enabled)
                now = datetime.utcnow()
                if add_late_events and random.random() < 0.2:
                    # 20% of events arrive 5 minutes late
                    event_time = now - timedelta(minutes=5)
                else:
                    event_time = now
                
                # Full message payload
                payload = {
                    "device_id": device_id,
                    "temperature": round(temperature, 2),
                    "humidity": round(humidity, 2),
                    "timestamp": event_time.isoformat(),
                    "location": f"Zone-{device_num % 3 + 1}",
                    "device_type": random.choice(["TypeX", "TypeY", "TypeZ"]),
                    "status": (
                        "CRITICAL" if temperature > 80 
                        else "WARN" if temperature > 70 
                        else "OK"
                    ),
                    "is_anomaly": is_anomaly
                }
                
                event_data_batch.add(EventData(json.dumps(payload)))
                messages_sent += 1
            
            # Send the batch
            producer.send_batch(event_data_batch)
            
            if messages_sent % (num_devices * 10) == 0:
                elapsed = time.time() - start_time
                print(f"  {messages_sent} messages sent ({elapsed:.1f}s)")
            
            time.sleep(1.0 / readings_per_second)
    
    print(f"\nSimulation complete: {messages_sent} messages sent")
    return messages_sent

# Start the simulation in a separate thread
import threading
producer_thread = threading.Thread(
    target=send_device_readings,
    kwargs={
        "num_devices": 5,
        "readings_per_second": 2,
        "duration_seconds": 120,
        "add_late_events": True,
        "anomaly_probability": 0.1
    }
)
producer_thread.start()
print("IoT producer started in background!")

Module 9. Spark Structured Streaming Consumer

9.1 Parse and Transform the Stream

from pyspark.sql import functions as F
from pyspark.sql.types import *

# IoT message schema
device_schema = StructType([
    StructField("device_id",   StringType(),  nullable=False),
    StructField("temperature", FloatType(),   nullable=True),
    StructField("humidity",    FloatType(),   nullable=True),
    StructField("timestamp",   StringType(),  nullable=True),
    StructField("location",    StringType(),  nullable=True),
    StructField("device_type", StringType(),  nullable=True),
    StructField("status",      StringType(),  nullable=True),
    StructField("is_anomaly",  BooleanType(), nullable=True),
])

# Read the raw stream from Event Hub
raw_df = spark.readStream \
    .format("eventhubs") \
    .options(**eh_conf) \
    .load()

# Transform binary body into structured JSON
parsed_df = (
    raw_df
    # Convert binary to string
    .withColumn("body_str", F.col("body").cast("string"))
    # Parse the JSON
    .withColumn("data", F.from_json("body_str", device_schema))
    # Extract JSON fields
    .select(
        "data.device_id",
        "data.temperature",
        "data.humidity",
        F.to_timestamp("data.timestamp").alias("event_time"),
        "data.location",
        "data.device_type",
        "data.status",
        "data.is_anomaly",
        # Event Hub metadata
        F.col("enqueuedTime").alias("enqueued_time"),
        F.col("partition"),
        F.col("offset")
    )
    # Filter invalid messages
    .filter(F.col("device_id").isNotNull())
    .filter(F.col("temperature").isNotNull())
)

print("IoT stream configured:")
parsed_df.printSchema()

9.2 Basic Stream Transformations

# Business transformations
enriched_stream = (
    parsed_df
    # Categorize temperature
    .withColumn(
        "temp_category",
        F.when(F.col("temperature") > 80, "CRITICAL")
         .when(F.col("temperature") > 70, "WARNING")
         .when(F.col("temperature") > 60, "ELEVATED")
         .otherwise("NORMAL")
    )
    # Calculate heat index (temperature + humidity factor)
    .withColumn(
        "heat_index",
        F.round(
            F.col("temperature") + 0.1 * F.col("humidity"),
            2
        )
    )
    # Add processing timestamp
    .withColumn("processing_time", F.current_timestamp())
    .withColumn("processing_delay_sec",
        (F.unix_timestamp("processing_time") - 
         F.unix_timestamp("event_time")).cast("double")
    )
)

# Write to a Delta table (streaming mode)
query_delta = (
    enriched_stream.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", 
            "abfss://checkpoints@monitoringdls.dfs.core.windows.net/device-stream/")
    .trigger(processingTime="30 seconds")
    .toTable("monitoringcatalog.streaming.device_readings")
)

print("Stream running...")
# query_delta.awaitTermination()  # Uncomment to block until stopped

Module 10. Schema Evolution and Late-Arriving Data

10.1 Handling Schema Evolution

# The producer added new fields "status" and "timestamp"
# Update the consumer schema

device_schema_v2 = StructType([
    StructField("device_id",   StringType(),  nullable=False),
    StructField("temperature", FloatType(),   nullable=True),
    StructField("humidity",    FloatType(),   nullable=True),
    StructField("timestamp",   StringType(),  nullable=True),  # NEW
    StructField("status",      StringType(),  nullable=True),   # NEW
    StructField("location",    StringType(),  nullable=True),
    StructField("device_type", StringType(),  nullable=True),
])

# from_json handles missing fields gracefully
# If an old message doesn't have "status" -> NULL value, no error!
parsed_v2 = (
    raw_df
    .withColumn("body_str", F.col("body").cast("string"))
    .withColumn("data", F.from_json("body_str", device_schema_v2))
    .select(
        "data.*",
        F.col("enqueuedTime").alias("enqueued_time")
    )
)

print("Schema v2 compatible with both old AND new messages!")

10.2 Watermarking — Handling Late Data

from pyspark.sql import functions as F

# Add the event time column
df_with_event_time = parsed_v2 \
    .withColumn(
        "event_timestamp",
        F.to_timestamp("timestamp", "yyyy-MM-dd'T'HH:mm:ss.SSSSSS")
    )

# Apply a 3-minute watermark
# This means: accept late data up to 3 minutes behind
watermarked_df = df_with_event_time.withWatermark("event_timestamp", "3 minutes")

print("Watermark configured: 3 minutes")
print("Spark will wait 3 minutes for late-arriving data")
print("Data arriving more than 3 minutes late will be IGNORED")

10.3 Impact of Watermark on Aggregations

timeline
    title 3-Minute Watermark — Behavior
    
    10:00 : Event A received
    10:01 : Event B received
    10:02m30 : Late Event C received
    10:03 : Late Event D received
    10:04 : Late Event E received

Module 11. Checkpointing and Fault Tolerance

11.1 Why Checkpointing is Essential

sequenceDiagram
    participant S as Spark Stream
    participant CP as Checkpoint\n(ADLS Gen2)
    participant EH as Event Hub
    participant DT as Delta Table

    S->>EH: Read offsets 1000-1500
    S->>DT: Write 500 events
    S->>CP: Save offset=1500, state, metadata
    
    Note over S: Cluster crash!
    
    S->>CP: Read last checkpoint
    CP-->>S: Resume from offset=1500
    S->>EH: Read offsets 1500-2000 (NOT 1000-2000!)
    S->>DT: Write 500 new events (NO duplicates!)
    
    Note over S,DT: Exactly-once processing\nthanks to checkpointing!

11.2 Checkpointing Configuration

import os

# Define checkpoint paths
checkpoint_base = "abfss://checkpoints@monitoringdls.dfs.core.windows.net"

# Query with full checkpointing
query_production = (
    watermarked_df
    .groupBy(
        F.window("event_timestamp", "1 minute"),
        "device_id"
    )
    .agg(
        F.avg("temperature").alias("avg_temp"),
        F.max("temperature").alias("max_temp"),
        F.count("*").alias("reading_count")
    )
    .writeStream
    .format("delta")
    .outputMode("update")
    .option("checkpointLocation", 
            f"{checkpoint_base}/device-aggregations/")
    .trigger(processingTime="1 minute")
    .toTable("monitoringcatalog.streaming.device_minutely_stats")
)

print("Stream with checkpoint started!")
print(f"Checkpoint: {checkpoint_base}/device-aggregations/")

# Inspect the checkpoint
files = dbutils.fs.ls(f"{checkpoint_base}/device-aggregations/")
for f in files:
    print(f"  {f.name}")
# -> offsets/ commits/ state/ metadata

11.3 Checkpoint Directory Contents

checkpoints/device-aggregations/
├── offsets/              <- Last Event Hub offsets read
│   ├── 0                 <- Checkpoint for version 0
│   ├── 1                 <- Checkpoint for version 1
│   └── ...
├── commits/              <- Successful commits
│   ├── 0
│   └── ...
├── state/                <- Stateful aggregation state
│   └── 0/
│       └── 0.delta       <- Temporal window state
└── metadata              <- Stream metadata

Module 12. Temporal Aggregations — Windowed Operations

12.1 Types of Temporal Windows

graph TB
    subgraph "Tumbling Window (Non-overlapping)"
        TW1[Window: 10:00-10:01\nAVG temp=65C]
        TW2[Window: 10:01-10:02\nAVG temp=68C]
        TW3[Window: 10:02-10:03\nAVG temp=72C]
        
        Events1[Events 10:00:00\n10:00:30\n10:00:45] --> TW1
        Events2[Events 10:01:05\n10:01:40] --> TW2
        Events3[Events 10:02:15\n10:02:55] --> TW3
    end
    
    subgraph "Sliding Window (Overlapping)"
        SW1[Window: 10:00-10:05\nAVG temp=66C]
        SW2[Window: 10:01-10:06\nAVG temp=68C]
        SW3[Window: 10:02-10:07\nAVG temp=71C]
        
        note[Slides every 1 minute\nOverlap: 4 minutes]
    end

12.2 Implementing Windows

from pyspark.sql import functions as F

# -- TUMBLING WINDOW: Average per minute -----------------------
tumbling_agg = (
    watermarked_df
    .groupBy(
        F.window("event_timestamp", "1 minute"),  # 1-minute window
        "device_id",
        "location"
    )
    .agg(
        F.avg("temperature").alias("avg_temperature"),
        F.max("temperature").alias("max_temperature"),
        F.min("temperature").alias("min_temperature"),
        F.avg("humidity").alias("avg_humidity"),
        F.count("*").alias("reading_count"),
        F.count(F.when(F.col("status") == "CRITICAL", 1)).alias("critical_count")
    )
    .select(
        F.col("window.start").alias("window_start"),
        F.col("window.end").alias("window_end"),
        "device_id",
        "location",
        "avg_temperature",
        "max_temperature",
        "min_temperature",
        "avg_humidity",
        "reading_count",
        "critical_count"
    )
)

# Write per-minute aggregations
tumbling_query = (
    tumbling_agg.writeStream
    .outputMode("update")  # Update completed windows
    .format("delta")
    .option("checkpointLocation",
            "abfss://checkpoints@monitoringdls.dfs.core.windows.net/tumbling-1min/")
    .trigger(processingTime="30 seconds")
    .toTable("monitoringcatalog.streaming.device_minutely_stats")
)

# -- SLIDING WINDOW: 5-minute trends --------------------------
sliding_agg = (
    watermarked_df
    .groupBy(
        # 5-minute window sliding every 1 minute
        F.window("event_timestamp", "5 minutes", "1 minute"),
        "device_id"
    )
    .agg(
        F.avg("temperature").alias("rolling_avg_5min"),
        F.percentile_approx("temperature", 0.95).alias("p95_temperature"),
        F.count("*").alias("reading_count_5min")
    )
)

print("Windowed aggregations configured!")

Module 13. Stateful Processing — Complex Alerts

13.1 State-Based Alert Detection

# Alert: Detect if a device exceeds 75 C three times in 1 minute
# This is STATEFUL processing because it must remember past events

from pyspark.sql import functions as F

# Read the stream and add an is_critical flag
critical_readings = (
    parsed_df
    .withColumn("event_time", F.to_timestamp("timestamp"))
    .withColumn(
        "is_critical",
        F.col("temperature") > 75
    )
)

# Apply watermark for late-arriving data
critical_with_wm = critical_readings.withWatermark("event_time", "2 minutes")

# Count critical readings per 1-minute window per device
alert_df = (
    critical_with_wm
    .groupBy(
        F.window("event_time", "1 minute"),  # 1-minute tumbling window
        "device_id",
        "location"
    )
    .agg(
        F.count(F.when(F.col("is_critical"), 1)).alias("critical_count"),
        F.avg("temperature").alias("avg_temp"),
        F.max("temperature").alias("max_temp")
    )
    .where(F.col("critical_count") >= 3)  # Alert if >= 3 critical readings
    .withColumn(
        "alert_message",
        F.concat(
            F.lit("ALERT: Device "),
            F.col("device_id"),
            F.lit(" at "),
            F.col("location"),
            F.lit(" - "),
            F.col("critical_count"),
            F.lit(" readings > 75C in 1 minute!")
        )
    )
)

# Write alerts to a dedicated Delta table
alert_query = (
    alert_df.writeStream
    .outputMode("update")
    .format("delta")
    .option("checkpointLocation",
            "abfss://checkpoints@monitoringdls.dfs.core.windows.net/alerts/")
    .trigger(processingTime="15 seconds")
    .toTable("monitoringcatalog.streaming.device_alerts")
)

print("Stateful alerting system started!")

13.2 Advanced Stateful Processing with foreachBatch

def process_alert_batch(batch_df, batch_id: int):
    """
    Process each micro-batch of alerts.
    Used for complex actions not possible via standard writeStream.
    """
    if batch_df.isEmpty():
        return
    
    # Count alerts in this batch
    alert_count = batch_df.count()
    print(f"Batch {batch_id}: {alert_count} alerts detected")
    
    # 1. Save to Delta Lake
    batch_df.write \
        .format("delta") \
        .mode("append") \
        .saveAsTable("monitoringcatalog.streaming.device_alerts")
    
    # 2. Send notifications for critical alerts
    critical_alerts = batch_df.filter(F.col("critical_count") >= 5).collect()
    
    for alert in critical_alerts:
        # Send a Teams/Email/PagerDuty alert
        send_notification(
            f"CRITICAL: {alert['device_id']} at {alert['location']}: "
            f"{alert['critical_count']} critical readings!"
        )
    
    # 3. Update monitoring statistics
    batch_df.write \
        .format("delta") \
        .mode("overwrite") \
        .option("replaceWhere", f"batch_id = {batch_id}") \
        .saveAsTable("monitoringcatalog.streaming.alert_monitoring")

def send_notification(message: str):
    """Send a notification (stub — implement with Teams webhook, etc.)"""
    print(f"NOTIFICATION: {message}")

# Start the stream with foreachBatch
alert_query_v2 = (
    alert_df.writeStream
    .outputMode("update")
    .foreachBatch(process_alert_batch)
    .option("checkpointLocation",
            "abfss://checkpoints@monitoringdls.dfs.core.windows.net/alerts-v2/")
    .trigger(processingTime="30 seconds")
    .start()
)

Module 14. Real-Time Anomaly Detection (ML)

14.1 Train and Register an Anomaly Model

import mlflow
import mlflow.sklearn
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from mlflow.models import infer_signature

mlflow.set_experiment("/Users/user@company.com/device-anomaly-detection")
mlflow.set_registry_uri("databricks-uc")

# Generate synthetic training data
np.random.seed(7)
n_normal = 10000

# Normal data: temperature 55-75 C, humidity 30-70%
normal_data = np.column_stack([
    np.random.normal(65, 5, n_normal),   # Normal temperature
    np.random.normal(50, 10, n_normal)   # Normal humidity
])

# A few anomalies: temperature > 80 C
anomalies = np.column_stack([
    np.random.uniform(80, 95, 100),
    np.random.uniform(10, 90, 100)
])

X_train = np.vstack([normal_data, anomalies])

# Normalize features
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)

# Train Isolation Forest
with mlflow.start_run(run_name="isolation_forest_device_anomaly"):
    
    model = IsolationForest(
        n_estimators=150,
        contamination=0.01,  # 1% expected anomalies
        random_state=7
    )
    model.fit(X_train_scaled)
    
    # Predict on training data
    predictions = model.predict(X_train_scaled)
    anomaly_rate = (predictions == -1).mean()
    
    mlflow.log_params({
        "n_estimators": 150,
        "contamination": 0.01,
        "n_features": 2
    })
    mlflow.log_metrics({
        "anomaly_rate": anomaly_rate,
        "n_training_samples": len(X_train)
    })
    
    # Infer model signature (required for Unity Catalog)
    signature = infer_signature(
        X_train_scaled,
        model.predict(X_train_scaled)
    )
    
    # Register the model in Unity Catalog
    result = mlflow.sklearn.log_model(
        sk_model=model,
        artifact_path="isolation-forest",
        signature=signature,
        registered_model_name="monitoringcatalog.streaming.device_anomaly_detector"
    )
    
    print(f"Model registered: {result.model_uri}")
    print(f"Anomaly rate: {anomaly_rate:.3f}")

14.2 Applying the ML Model to the Stream in Real Time

import mlflow
import pandas as pd
from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType

# Load the model via the "production" alias
# (configure alias in Unity Catalog after validation)
model_uri = "models:/monitoringcatalog.streaming.device_anomaly_detector@production"

# Load the model once (not per micro-batch)
loaded_model = mlflow.sklearn.load_model(model_uri)
scaler_model = StandardScaler()  # In real implementation, load the saved scaler

# Broadcast the model to all Spark workers
broadcast_model = sc.broadcast(loaded_model)
broadcast_scaler = sc.broadcast(scaler_model)

# UDF for anomaly detection
@F.pandas_udf(IntegerType())
def detect_anomaly(temperature: pd.Series, humidity: pd.Series) -> pd.Series:
    """
    Vectorized UDF to detect anomalies.
    Returns: 1 = normal, -1 = anomaly
    """
    model = broadcast_model.value
    
    # Build the feature matrix
    features = np.column_stack([
        temperature.fillna(65).values,
        humidity.fillna(50).values
    ])
    
    # Predict
    predictions = model.predict(features)  # 1 = normal, -1 = anomaly
    
    return pd.Series(predictions)

# Apply anomaly detection to the stream
anomaly_stream = (
    parsed_df
    .withColumn("event_time", F.to_timestamp("timestamp"))
    .withColumn(
        "ml_anomaly_score",
        detect_anomaly(F.col("temperature"), F.col("humidity"))
    )
    .withColumn(
        "is_ml_anomaly",
        F.col("ml_anomaly_score") == -1
    )
    .withColumn(
        "anomaly_type",
        F.when(
            (F.col("temperature") > 80) & F.col("is_ml_anomaly"),
            "HIGH_TEMPERATURE_ANOMALY"
        ).when(
            (F.col("humidity") > 85) & F.col("is_ml_anomaly"),
            "HIGH_HUMIDITY_ANOMALY"
        ).when(
            F.col("is_ml_anomaly"),
            "MULTIVARIATE_ANOMALY"
        ).otherwise("NORMAL")
    )
)

# Write results with ML detection
anomaly_query = (
    anomaly_stream.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation",
            "abfss://checkpoints@monitoringdls.dfs.core.windows.net/ml-anomaly/")
    .trigger(processingTime="15 seconds")
    .toTable("monitoringcatalog.streaming.device_ml_anomalies")
)

print("ML anomaly detection started!")

Module 15. Data Stream Enrichment

15.1 Stream-Static Join — Enrich with Reference Data

# Static reference data (device metadata)
device_metadata = spark.createDataFrame([
    ("device-001", "Building-A", "Floor-3", "TypeX", "HVAC"),
    ("device-002", "Building-A", "Floor-5", "TypeY", "Server Room"),
    ("device-003", "Building-B", "Floor-1", "TypeX", "Factory Floor"),
    ("device-004", "Building-B", "Floor-2", "TypeZ", "Clean Room"),
    ("device-005", "Building-C", "Floor-1", "TypeY", "Data Center"),
], ["device_id", "building", "floor", "device_type", "zone"])

# Cache reference data for repeated joins
device_metadata.cache()

# Stream-Static Join (real-time enrichment)
enriched_stream = (
    parsed_df.alias("stream")
    .join(
        device_metadata.alias("meta"),
        on="device_id",
        how="left"  # left join: keep all events even without metadata
    )
    .select(
        "stream.device_id",
        "stream.temperature",
        "stream.humidity",
        "stream.timestamp",
        "stream.status",
        # Enriched fields from the metadata table
        "meta.building",
        "meta.floor",
        "meta.zone",
        # Detect unknown devices
        F.when(F.col("meta.device_id").isNull(), True)
         .otherwise(False)
         .alias("is_unknown_device")
    )
)

# Example of enriched output
# device-001 | 67.5 C | 45% | Building-A | Floor-3 | HVAC | False
# unknown-x  | 72.0 C | 60% | null       | null    | null | True

print("Stream enriched with device metadata!")
enriched_stream.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", 
            "abfss://checkpoints@monitoringdls.dfs.core.windows.net/enriched/") \
    .toTable("monitoringcatalog.streaming.device_enriched") \
    .start()

Module 16. Autoscaling for Streaming Workloads

16.1 Streaming Cluster Configuration

// Optimized configuration for streaming workloads
{
  "cluster_name": "streaming-prod-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": 10
  },
  
  "spark_conf": {
    "spark.streaming.backpressure.enabled": "true",
    "spark.sql.shuffle.partitions": "200",
    "spark.databricks.delta.optimizeWrite.enabled": "true"
  },
  
  "autotermination_minutes": 0
}

16.2 Autoscaling + Checkpointing: The Perfect Combination

graph LR
    subgraph "Normal data flow"
        EH[Event Hub\n1000 events/min]
        Cluster2[Cluster\n2 workers]
        Delta[Delta Table]
        EH --> Cluster2 --> Delta
    end
    
    subgraph "Traffic spike"
        EH2[Event Hub\n50000 events/min]
        Cluster10[Cluster\n10 workers\nAutoScale up]
        Delta2[Delta Table]
        CP[Checkpoint\nState preserved]
        EH2 --> Cluster10 --> Delta2
        Cluster10 --> CP
    end
    
    subgraph "After the spike"
        EH3[Event Hub\n1000 events/min]
        Cluster3[Cluster\n3 workers\nAutoScale down]
        CP2[Checkpoint\nState resumed]
        EH3 --> Cluster3
        CP2 --> Cluster3
    end
    
    note[No data loss\nthanks to checkpointing\nOptimal cost\nthanks to autoscaling]

Module 17. Monitoring Streaming Jobs

17.1 Spark UI Metrics for Streaming

# Start a named stream to simplify monitoring
monitoring_query = (
    enriched_stream.writeStream
    .format("memory")  # Write to memory for inspection
    .queryName("device_enriched_monitoring")  # Stream name
    .outputMode("append")
    .trigger(processingTime="5 seconds")
    .start()
)

# Access stream metrics
import time
time.sleep(30)  # Wait for a few batches

# Stream statistics
progress = monitoring_query.lastProgress
if progress:
    print("=== Stream Metrics ===")
    print(f"Batch ID: {progress['batchId']}")
    print(f"Rows received: {progress['numInputRows']}")
    print(f"Rows/second (input): {progress['inputRowsPerSecond']:.2f}")
    print(f"Rows/second (processing): {progress['processedRowsPerSecond']:.2f}")
    print(f"Batch duration: {progress['batchDuration']}ms")
    print(f"Sources: {progress['sources']}")

# Check if the stream is active
print(f"\nStream active: {monitoring_query.isActive}")
print(f"Stream status: {monitoring_query.status}")

17.2 Key Metrics to Monitor

17.3 Monitoring Dashboard

# Create an automated monitoring dashboard
def streaming_health_check() -> dict:
    """Check the health of active streams."""
    
    health = {"timestamp": datetime.now().isoformat(), "streams": []}
    
    for query in spark.streams.active:
        progress = query.lastProgress
        
        if progress:
            input_rate = progress.get("inputRowsPerSecond", 0)
            proc_rate = progress.get("processedRowsPerSecond", 0)
            batch_dur = progress.get("batchDuration", 0)
            
            # Determine health status
            if proc_rate >= input_rate * 0.9:
                status = "HEALTHY"
            elif proc_rate >= input_rate * 0.7:
                status = "DEGRADED"
            else:
                status = "CRITICAL"
            
            stream_health = {
                "name": query.name,
                "status": status,
                "input_rate": round(input_rate, 2),
                "processing_rate": round(proc_rate, 2),
                "batch_duration_ms": batch_dur,
                "is_active": query.isActive
            }
            health["streams"].append(stream_health)
    
    return health

# Real-time check
health = streaming_health_check()
print(f"=== Spark Stream Health ===")
for stream in health["streams"]:
    print(f"\n[{stream['status']}] {stream['name']}")
    print(f"  Input: {stream['input_rate']} rows/s")
    print(f"  Processing: {stream['processing_rate']} rows/s")
    print(f"  Batch duration: {stream['batch_duration_ms']}ms")

Module 18. Troubleshooting Latency Issues

18.1 Data Skew in Streaming

# PROBLEM: 95% of data comes from device-001
# -> A single executor handles 95% of the load -> LATENCY

# DIAGNOSIS: Analyze partition distribution
skewed_df = parsed_df.groupBy("device_id").count()
skewed_df.show(20, truncate=False)
# device-001 | 95000  <- SKEW DETECTED!
# device-002 | 2500
# device-003 | 1500
# device-004 | 1000

# SOLUTION 1: Repartition after filtering
balanced_df = parsed_df.repartition(10, "device_id")

# SOLUTION 2: Salting to better distribute skewed keys
import pyspark.sql.functions as F

df_salted = parsed_df.withColumn(
    "salted_device_id",
    F.concat(
        F.col("device_id"),
        F.lit("_"),
        (F.rand() * 5).cast("int").cast("string")
    )
)

# SOLUTION 3: Increase spark.sql.shuffle.partitions
spark.conf.set("spark.sql.shuffle.partitions", "400")

# SOLUTION 4: AQE (Adaptive Query Execution) to handle skew automatically
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "3.0")

print("Anti-skew measures applied!")

18.2 Streaming Troubleshooting Checklist

Module 19. Streaming Cost Optimization

19.1 Cost Optimization Strategies

# STRATEGY 1: Adaptive trigger interval
# Reduce frequency when data is sparse
economy_query = (
    parsed_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", 
            "abfss://checkpoints@monitoringdls.dfs.core.windows.net/economy/")
    
    # Instead of processingTime='1 second' (360 batches/hour)
    # Use processingTime='30 seconds' (120 batches/hour) -> 3x cheaper!
    .trigger(processingTime="30 seconds")
    
    .toTable("monitoringcatalog.streaming.device_economic")
)

# STRATEGY 2: availableNow for incremental batch
# If < 1 minute of latency is acceptable
batch_incremental = (
    parsed_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", 
            "abfss://checkpoints@monitoringdls.dfs.core.windows.net/batch-incr/")
    
    # Process everything available, then STOP
    # -> Zero cost between executions
    .trigger(availableNow=True)
    
    .toTable("monitoringcatalog.streaming.device_batch_incremental")
    .start()
)

batch_incremental.awaitTermination()  # Wait for completion
print("Incremental batch complete — cluster can now stop!")

19.2 Cost Comparison by Strategy

Module 20. Real-Time Reference Architecture

20.1 Lambda / Kappa Architecture for Databricks

graph TB
    subgraph "Sources"
        IOT2[IoT Devices\nSensors]
        APP2[Applications\nMicroservices]
        LOGS2[System Logs\nDiagnostics]
    end
    
    subgraph "Ingestion"
        EH3[Azure Event Hub\nReal-time ingestion]
        EH3B[Azure Event Hub\nCapture -> ADLS Raw]
    end
    
    subgraph "Processing Layer"
        subgraph "Speed Layer (Streaming)"
            SS[Spark Structured Streaming\n1-5 min windows]
            ANO[Anomaly Detection\nReal-time ML]
            ALE[Alert Engine\nStateful Processing]
        end
        
        subgraph "Batch Layer (Lambda)"
            BATCH[Spark Batch\nHistorical computations]
            HIST[Historical Analytics\nDays/Weeks]
        end
    end
    
    subgraph "Storage (Delta Lake)"
        RAW[Bronze: Raw Events]
        SILVER2[Silver: Cleaned Stream]
        GOLD2[Gold: Aggregations\n1min / 5min / 1h]
        ALERTS[Alerts Table\nAnomalies]
    end
    
    subgraph "Consumption"
        PBI2[Power BI\nReal-time Dashboard]
        API3[REST API\nExternal Alerts]
        ML2[ML Models\nTraining]
    end
    
    IOT2 & APP2 & LOGS2 --> EH3
    EH3 --> SS & ANO & ALE
    EH3B --> RAW
    SS --> SILVER2 --> GOLD2
    ANO & ALE --> ALERTS
    BATCH --> GOLD2
    RAW --> BATCH
    GOLD2 --> PBI2 & API3 & ML2
    ALERTS --> API3

20.2 Delta Live Tables (DLT) Configuration

# Delta Live Tables — declarative pipeline for streaming
import dlt
from pyspark.sql import functions as F
from pyspark.sql.types import *

# Bronze: Raw ingestion from Event Hub
@dlt.table(
    name="device_events_bronze",
    comment="Raw IoT events from Azure Event Hub",
    table_properties={"quality": "bronze"}
)
def device_bronze():
    return (
        spark.readStream
        .format("eventhubs")
        .options(**eh_conf)
        .load()
        .withColumn("body_str", F.col("body").cast("string"))
        .withColumn("ingestion_time", F.current_timestamp())
    )

# Silver: Parsing and validation
@dlt.table(
    name="device_events_silver",
    comment="Cleaned and validated IoT events",
    table_properties={"quality": "silver"}
)
@dlt.expect("valid_temperature", "temperature > 0 AND temperature < 200")
@dlt.expect("valid_humidity", "humidity >= 0 AND humidity <= 100")
def device_silver():
    return (
        dlt.read_stream("device_events_bronze")
        .withColumn("data", F.from_json("body_str", device_schema))
        .select("data.*", "ingestion_time")
        .withColumn("event_time", F.to_timestamp("timestamp"))
        .withWatermark("event_time", "5 minutes")
    )

# Gold: Per-minute aggregations
@dlt.table(
    name="device_stats_1min_gold",
    comment="IoT statistics per device per minute",
    table_properties={"quality": "gold"}
)
def device_gold_1min():
    return (
        dlt.read_stream("device_events_silver")
        .groupBy(
            F.window("event_time", "1 minute"),
            "device_id",
            "location"
        )
        .agg(
            F.avg("temperature").alias("avg_temp"),
            F.max("temperature").alias("max_temp"),
            F.count("*").alias("reading_count"),
            F.count(F.when(F.col("temperature") > 75, 1)).alias("alert_count")
        )
    )

Module 21. Summary and Best Practices

21.1 Production Streaming Pipeline Checklist

mindmap
  root((Streaming\nProduction Ready))
    Reliability
      Checkpointing always enabled
      Watermark for late events
      Error handling with foreachBatch
      Alerts on stream status
    Performance
      Adaptive trigger interval
      Autoscaling enabled
      Avoid slow Python UDFs
      Broadcast small tables
    Costs
      Trigger > 30s when possible
      Spot instances for non-critical
      availableNow for incremental batch
      Monitor DBU costs
    Security
      Secrets in Key Vault
      Dedicated Consumer Groups
      TLS for Event Hub
      Audit logging enabled
    Quality
      Schema defined manually
      Data quality with DLT expect
      Monitor Processing Rate
      Tests with synthetic data

21.2 Recommended Patterns

Module 22. Glossary

Search Terms

real-time · data · processing · azure · databricks · spark · engineering · analytics · streaming · event · hub · stream · architecture · checkpointing · configuration · comparison · cost · model · aggregations · anomaly · autoscaling · cases · checklist · cli