Real-Time Anomaly Detection with Structured Streaming and Delta Lake

Table of Contents

Databricks Challenge - This article is part of a series.

Part : Data Quality Gates: Lakehouse Monitoring and DLT Expectations

Part : Delta Sharing: Zero-Copy Data Exchange Across Workspaces and Beyond

Part : Multi-Tenant Data Platform: Row Filters, Column Masks, and ABAC on Unity Catalog

Part : This Article

Part : SCD Type 2 Done Right: Declarative Pipelines and Change Data Feed

Your IoT sensors are streaming thousands of readings per second. Somewhere in that flood, a compressor is overheating, a valve is stuck, a coolant line is losing pressure. By the time someone notices in a batch report tomorrow morning, the damage is done.

This is the kind of problem that Structured Streaming on Databricks was built for. Not the toy wordcount demos from conference talks – actual stateful processing that tracks per-device baselines and fires alerts the moment something drifts outside normal range.

In this challenge, we’ll build a complete anomaly detection pipeline: simulated IoT ingestion, stateful z-score computation per device, alert deduplication with Delta merge, and pipeline monitoring using system tables. Everything runs on Databricks, everything is production-grade, and you can follow along in a notebook.

The Problem
#

You have hundreds of IoT sensors reporting temperature, pressure, and vibration metrics. Normal operating ranges vary per device – a compressor runs hotter than an air handler. Hardcoded thresholds don’t work because each device has its own baseline.

What you need is a per-device rolling statistical model that adapts to each sensor’s normal behavior and flags deviations in real time. A rolling z-score does exactly this: it tracks the mean and standard deviation over a window and flags any reading that falls outside a configurable number of standard deviations.

Architecture
#

The pipeline has four stages:

flowchart LR
    subgraph Ingest["1. Ingestion"]
        K["Simulated Sensor Stream
(Auto Loader / Kafka)"]
    end

    subgraph Process["2. Stateful Processing"]
        S["mapGroupsWithState
Rolling z-score per device"]
    end

    subgraph Sink["3. Alert Sink"]
        D["Delta Table
anomaly_alerts"]
    end

    subgraph Monitor["4. Monitoring"]
        M["System Tables
streaming.query_progress"]
    end

    K --> S --> D --> M

    style K fill:#3498db,color:white
    style S fill:#e74c3c,color:white
    style D fill:#2ecc71,color:white
    style M fill:#f39c12,color:white

Step 1: Simulate the Sensor Stream
#

In production you’d read from Kafka or an Event Hub. For this challenge, we generate synthetic data using Auto Loader on a landing volume. This gives us the same readStream interface without external dependencies.

First, create the landing zone and target tables:

1
2
3
4


CREATE CATALOG IF NOT EXISTS iot_demo;
CREATE SCHEMA IF NOT EXISTS iot_demo.streaming;

CREATE VOLUME IF NOT EXISTS iot_demo.streaming.sensor_landing;

Now generate some synthetic sensor data. We’ll write JSON files to the volume to simulate arriving batches:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50


import json
import random
import time
from datetime import datetime, timezone

DEVICES = [f"sensor_{i:03d}" for i in range(20)]
METRICS = ["temperature", "pressure", "vibration"]

def generate_batch(n_records=100, anomaly_pct=0.05):
    """Generate a batch of sensor readings.
    A small percentage are anomalous (value shifted by 4-6 sigma)."""
    records = []
    for _ in range(n_records):
        device = random.choice(DEVICES)
        metric = random.choice(METRICS)

        # Normal baselines per metric type
        baselines = {
            "temperature": (72.0, 3.0),
            "pressure": (14.7, 0.5),
            "vibration": (0.02, 0.005),
        }
        mean, std = baselines[metric]

        if random.random() < anomaly_pct:
            # Inject anomaly: shift value 4-6 sigma away
            shift = random.choice([-1, 1]) * random.uniform(4, 6) * std
            value = mean + shift
        else:
            value = random.gauss(mean, std)

        records.append({
            "device_id": device,
            "metric": metric,
            "value": round(value, 4),
            "event_time": datetime.now(timezone.utc).isoformat(),
        })
    return records

# Write 10 batches of 100 records each
volume_path = "/Volumes/iot_demo/streaming/sensor_landing"
for batch_num in range(10):
    records = generate_batch(100, anomaly_pct=0.05)
    file_path = f"{volume_path}/batch_{batch_num:04d}.json"
    with open(file_path, "w") as f:
        for r in records:
            f.write(json.dumps(r) + "\n")
    time.sleep(0.5)

print(f"Wrote 10 batches to {volume_path}")

Step 2: Ingest with Auto Loader
#

Auto Loader (cloudFiles) picks up new files as they land. We parse the JSON and add a processing timestamp:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24


from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, TimestampType
)

sensor_schema = StructType([
    StructField("device_id", StringType(), False),
    StructField("metric", StringType(), False),
    StructField("value", DoubleType(), False),
    StructField("event_time", StringType(), False),
])

raw_stream = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation",
            "/Volumes/iot_demo/streaming/sensor_landing/_schema")
    .schema(sensor_schema)
    .load("/Volumes/iot_demo/streaming/sensor_landing/")
    .withColumn("event_time",
                F.to_timestamp("event_time"))
    .withWatermark("event_time", "5 minutes")
)

The watermark is important. It tells Spark “I don’t expect data older than 5 minutes” which allows it to prune state for devices that stop reporting. Without a watermark, state grows unbounded and your checkpoint balloons.

Step 3: Stateful Anomaly Detection with `mapGroupsWithState`
#

This is where the real work happens. For each device+metric combination, we maintain a rolling window of the last N readings. On every micro-batch, we compute the mean and standard deviation, then check if the new reading is anomalous.

Define the State
#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41


from dataclasses import dataclass
from typing import List

@dataclass
class DeviceState:
    """Rolling window state for one device+metric pair."""
    readings: List[float]  # last N values
    count: int
    sum_val: float
    sum_sq: float

    def add(self, value: float, window_size: int = 100) -> "DeviceState":
        self.readings.append(value)
        self.count += 1
        self.sum_val += value
        self.sum_sq += value * value

        # Evict oldest if window exceeded
        while len(self.readings) > window_size:
            old = self.readings.pop(0)
            self.count -= 1
            self.sum_val -= old
            self.sum_sq -= old * old

        return self

    @property
    def mean(self) -> float:
        return self.sum_val / self.count if self.count > 0 else 0.0

    @property
    def std(self) -> float:
        if self.count < 2:
            return 0.0
        variance = (self.sum_sq / self.count) - (self.mean ** 2)
        return max(variance, 0.0) ** 0.5

    def z_score(self, value: float) -> float:
        if self.std == 0:
            return 0.0
        return (value - self.mean) / self.std

The State Update Function
#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48


from pyspark.sql.streaming import GroupState, GroupStateTimeout
import json

WINDOW_SIZE = 100
Z_THRESHOLD = 3.5

def detect_anomalies(key, readings, state: GroupState):
    """
    Called per (device_id, metric) group in each micro-batch.
    Maintains rolling stats in state, emits alerts for anomalous readings.
    """
    device_id, metric = key

    # Recover or initialize state
    if state.exists:
        device_state = DeviceState(**json.loads(state.get))
    else:
        device_state = DeviceState(readings=[], count=0,
                                   sum_val=0.0, sum_sq=0.0)

    alerts = []
    for row in readings:
        value = row.value
        event_time = row.event_time

        z = device_state.z_score(value)
        device_state.add(value, WINDOW_SIZE)

        # Only flag after we have enough data for meaningful stats
        if device_state.count >= 10 and abs(z) > Z_THRESHOLD:
            alerts.append((
                device_id,
                metric,
                value,
                round(z, 3),
                device_state.mean,
                device_state.std,
                event_time,
                f"{device_id}_{metric}_{event_time}",  # alert_id
            ))

    # Persist state
    state.update(json.dumps(device_state.__dict__))

    # Set timeout so stale devices get cleaned up
    state.setTimeoutDuration("30 minutes")

    return iter(alerts)

Wire It Up
#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26


from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType,
    FloatType, TimestampType
)

alert_schema = StructType([
    StructField("device_id", StringType()),
    StructField("metric", StringType()),
    StructField("value", DoubleType()),
    StructField("z_score", FloatType()),
    StructField("rolling_mean", DoubleType()),
    StructField("rolling_std", DoubleType()),
    StructField("event_time", TimestampType()),
    StructField("alert_id", StringType()),
])

alerts_stream = (
    raw_stream
    .groupBy("device_id", "metric")
    .mapGroupsWithState(
        detect_anomalies,
        outputStructType=alert_schema,
        stateStructType=StringType(),
        timeoutConf=GroupStateTimeout.ProcessingTimeTimeout,
    )
)

Why mapGroupsWithState instead of window aggregations? Window-based approaches (tumbling, sliding) compute stats over fixed time intervals. But anomaly detection needs a per-device rolling history that persists across windows. mapGroupsWithState gives you full control over what’s stored, how long it lives, and when it’s evicted.

Step 4: Write Alerts to Delta with Deduplication
#

Anomalies can span multiple micro-batches (a sensor stuck at a high value keeps triggering). We use MERGE to deduplicate:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19


# Create the target table
spark.sql("""
    CREATE TABLE IF NOT EXISTS iot_demo.streaming.anomaly_alerts (
        alert_id STRING,
        device_id STRING,
        metric STRING,
        value DOUBLE,
        z_score FLOAT,
        rolling_mean DOUBLE,
        rolling_std DOUBLE,
        event_time TIMESTAMP,
        detected_at TIMESTAMP
    )
    USING DELTA
    PARTITIONED BY (metric)
    TBLPROPERTIES (
        'delta.enableChangeDataFeed' = 'true'
    )
""")

Use foreachBatch to run the merge:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29


from delta.tables import DeltaTable
from pyspark.sql import functions as F

def upsert_alerts(batch_df, batch_id):
    if batch_df.isEmpty():
        return

    batch_with_ts = batch_df.withColumn(
        "detected_at", F.current_timestamp()
    )

    target = DeltaTable.forName(spark,
                                "iot_demo.streaming.anomaly_alerts")

    (target.alias("t")
     .merge(batch_with_ts.alias("s"), "t.alert_id = s.alert_id")
     .whenNotMatchedInsertAll()
     .execute())

alerts_query = (
    alerts_stream
    .writeStream
    .foreachBatch(upsert_alerts)
    .option("checkpointLocation",
            "/Volumes/iot_demo/streaming/sensor_landing/_checkpoint/alerts")
    .outputMode("update")
    .trigger(processingTime="30 seconds")
    .start()
)

Enable Change Data Feed on the alerts table. Downstream consumers (dashboards, notification pipelines) can use CDF to read only new/changed alerts instead of scanning the full table every time. This is how you chain streaming pipelines efficiently.

Step 5: Monitor the Pipeline
#

A streaming pipeline that nobody watches is a streaming pipeline that silently fails. Databricks system tables give you observability without bolting on external monitoring.

Query Progress Metrics
#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


-- Pipeline health: processing rate, batch duration, state size
SELECT
    stream_name,
    timestamp,
    num_input_rows,
    input_rows_per_second,
    processed_rows_per_second,
    batch_duration_ms
FROM system.runtime.streaming_query_progress
WHERE stream_name LIKE '%anomaly%'
ORDER BY timestamp DESC
LIMIT 20

State Store Metrics
#

State store growth is the silent killer of streaming pipelines. If your state grows linearly, eventually checkpoint recovery takes longer than the batch interval and the pipeline falls behind permanently.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


-- State store size over time
SELECT
    timestamp,
    state_operators[0].num_rows_total AS state_rows,
    state_operators[0].memory_used_bytes / 1024 / 1024 AS state_mb
FROM system.runtime.streaming_query_progress
WHERE stream_name LIKE '%anomaly%'
    AND state_operators IS NOT NULL
ORDER BY timestamp DESC
LIMIT 50

Watch for these signals:

state_rows growing without bound: your timeout or eviction is broken
batch_duration_ms steadily increasing: state is too large for the configured compute
input_rows_per_second » processed_rows_per_second: backpressure, you’re falling behind

Alerting on Pipeline Lag
#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


# Quick check: is the pipeline keeping up?
from pyspark.sql import functions as F

lag_check = (
    spark.sql("""
        SELECT
            max(timestamp) AS last_progress,
            current_timestamp() - max(timestamp) AS lag
        FROM system.runtime.streaming_query_progress
        WHERE stream_name LIKE '%anomaly%'
    """)
)
lag_check.show()

If lag exceeds your trigger interval by more than 2x, investigate. Common causes: state store bloat, insufficient compute, upstream data skew (one device flooding the stream).

Gotchas and Lessons Learned
#

State store sizing. Each device+metric pair stores a rolling window of 100 floats. With 20 devices and 3 metrics, that’s 60 state entries – trivial. With 100,000 devices, you’re looking at 300,000 state entries and checkpoint recovery becomes a factor. Test with realistic cardinality before going to production.

Checkpoint recovery after schema evolution. If you change the state schema (say, adding a field to DeviceState), existing checkpoints won’t deserialize correctly. You have two options: (1) start fresh with a new checkpoint location, losing in-flight state, or (2) implement versioned serialization in your state class. Option 2 is more work but avoids the cold-start problem.

Late-arriving data. The watermark says “I’ll wait 5 minutes for stragglers.” Data arriving after the watermark is silently dropped. If your IoT devices have unreliable connectivity, increase the watermark – but understand the trade-off: higher watermark = more state = slower checkpoints. There’s no free lunch.

The z-score cold start. With fewer than 10 readings, the standard deviation is unreliable. The code above skips anomaly flagging until count >= 10. Depending on your ingest rate, this means the first few minutes after pipeline start produce no alerts. That’s intentional, not a bug.

Processing-time vs. event-time timeouts. We use ProcessingTimeTimeout because it’s simpler. But if your pipeline has variable micro-batch intervals (some batches take 5s, others 60s), event-time timeouts give more predictable device eviction. The trade-off is complexity in managing watermark alignment.

Putting It All Together
#

Here’s the complete pipeline in one notebook-ready block, minus the data generation:

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133


# Full pipeline - copy into a Databricks notebook

from pyspark.sql import functions as F
from pyspark.sql.types import *
from pyspark.sql.streaming import GroupState, GroupStateTimeout
from delta.tables import DeltaTable
from dataclasses import dataclass
from typing import List
import json

# --- Config ---
CATALOG = "iot_demo"
SCHEMA = "streaming"
VOLUME_PATH = f"/Volumes/{CATALOG}/{SCHEMA}/sensor_landing"
CHECKPOINT_PATH = f"{VOLUME_PATH}/_checkpoint/alerts"
WINDOW_SIZE = 100
Z_THRESHOLD = 3.5

# --- State class ---
@dataclass
class DeviceState:
    readings: List[float]
    count: int
    sum_val: float
    sum_sq: float

    def add(self, value, window_size=100):
        self.readings.append(value)
        self.count += 1
        self.sum_val += value
        self.sum_sq += value * value
        while len(self.readings) > window_size:
            old = self.readings.pop(0)
            self.count -= 1
            self.sum_val -= old
            self.sum_sq -= old * old
        return self

    @property
    def mean(self):
        return self.sum_val / self.count if self.count > 0 else 0.0

    @property
    def std(self):
        if self.count < 2:
            return 0.0
        variance = (self.sum_sq / self.count) - (self.mean ** 2)
        return max(variance, 0.0) ** 0.5

    def z_score(self, value):
        return (value - self.mean) / self.std if self.std > 0 else 0.0

# --- State function ---
def detect_anomalies(key, readings, state: GroupState):
    device_id, metric = key
    ds = DeviceState(**json.loads(state.get)) if state.exists else \
         DeviceState([], 0, 0.0, 0.0)

    alerts = []
    for row in readings:
        z = ds.z_score(row.value)
        ds.add(row.value, WINDOW_SIZE)
        if ds.count >= 10 and abs(z) > Z_THRESHOLD:
            alerts.append((
                device_id, metric, row.value, round(z, 3),
                ds.mean, ds.std, row.event_time,
                f"{device_id}_{metric}_{row.event_time}",
            ))

    state.update(json.dumps(ds.__dict__))
    state.setTimeoutDuration("30 minutes")
    return iter(alerts)

# --- Schemas ---
sensor_schema = StructType([
    StructField("device_id", StringType()),
    StructField("metric", StringType()),
    StructField("value", DoubleType()),
    StructField("event_time", StringType()),
])

alert_schema = StructType([
    StructField("device_id", StringType()),
    StructField("metric", StringType()),
    StructField("value", DoubleType()),
    StructField("z_score", FloatType()),
    StructField("rolling_mean", DoubleType()),
    StructField("rolling_std", DoubleType()),
    StructField("event_time", TimestampType()),
    StructField("alert_id", StringType()),
])

# --- Stream ---
raw = (
    spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", f"{VOLUME_PATH}/_schema")
    .schema(sensor_schema)
    .load(VOLUME_PATH)
    .withColumn("event_time", F.to_timestamp("event_time"))
    .withWatermark("event_time", "5 minutes")
)

alerts = (
    raw.groupBy("device_id", "metric")
    .mapGroupsWithState(
        detect_anomalies,
        outputStructType=alert_schema,
        stateStructType=StringType(),
        timeoutConf=GroupStateTimeout.ProcessingTimeTimeout,
    )
)

# --- Sink with MERGE ---
def upsert(batch_df, batch_id):
    if batch_df.isEmpty():
        return
    batch = batch_df.withColumn("detected_at", F.current_timestamp())
    target = DeltaTable.forName(spark,
                                f"{CATALOG}.{SCHEMA}.anomaly_alerts")
    (target.alias("t")
     .merge(batch.alias("s"), "t.alert_id = s.alert_id")
     .whenNotMatchedInsertAll()
     .execute())

query = (
    alerts.writeStream
    .foreachBatch(upsert)
    .option("checkpointLocation", CHECKPOINT_PATH)
    .outputMode("update")
    .trigger(processingTime="30 seconds")
    .start()
)

Conclusion
#

A streaming anomaly detection pipeline on Databricks doesn’t need external ML services, complex model training, or a Lambda architecture with separate batch and speed layers. Structured Streaming with mapGroupsWithState gives you everything: per-device statefulness, automatic scaling, exactly-once guarantees with Delta, and observability through system tables.

The rolling z-score approach works because most industrial sensors follow roughly normal distributions during normal operation. When they don’t, the z-score catches it fast. For more complex patterns (seasonal variation, multivariate correlation), you’d extend the state class – but the streaming skeleton stays the same.

Start with synthetic data, validate your thresholds, then swap the source for your real Kafka topic or cloud storage landing zone. The pipeline doesn’t change. That’s the point.

Author

Mariusz Derela

Cyber Security Specialist | DevSecOps | AI/ML

Databricks Challenge - This article is part of a series.

Part : Data Quality Gates: Lakehouse Monitoring and DLT Expectations

Part : Delta Sharing: Zero-Copy Data Exchange Across Workspaces and Beyond

Part : Multi-Tenant Data Platform: Row Filters, Column Masks, and ABAC on Unity Catalog

Part : This Article

Part : SCD Type 2 Done Right: Declarative Pipelines and Change Data Feed

The Problem #

Architecture #

Step 1: Simulate the Sensor Stream #

Step 2: Ingest with Auto Loader #

Step 3: Stateful Anomaly Detection with mapGroupsWithState #

Define the State #

The State Update Function #

Wire It Up #

Step 4: Write Alerts to Delta with Deduplication #

Step 5: Monitor the Pipeline #

Query Progress Metrics #

State Store Metrics #

Alerting on Pipeline Lag #

Gotchas and Lessons Learned #

Putting It All Together #

Conclusion #

Related