IoT Temperature Alerting and Metrics Aggregation

# IoT Temperature Alerting and Metrics Aggregation

# Case Overview

In IoT temperature monitoring, two kinds of requirements usually coexist: instantaneous per-reading alerting (every reading must be evaluated and dispatched on arrival) and rolling per-device metric aggregation (per-minute avg/extremes for dashboards and anomaly detection). These map onto StreamSQL's two operating modes: a non-aggregation synchronous transform (EmitSync) and a windowed asynchronous aggregation (Emit + AddSink).

This case is built around a single temperature reading stream stream and chains together five sub-scenarios:

Sub-scenario	Mode	Window	Key capability
A. Single-point temperature alert	Non-aggregation (sync)	none	`EmitSync` + `CASE WHEN` instant transform/filter
B. Rolling 1-minute device metrics	Event-time aggregation	`TumblingWindow('1m')`	Watermark, out-of-order tolerance, idle advance
C. High-cardinality device anomaly detection	Processing-time aggregation	`CountingWindow(50)`	`STATETTL` dead-key reaping
D. Deterministic test	Trigger	`TumblingWindow('5s')`	`TriggerWindow()` immediate window flush
E. Global-window condition-driven alert	Processing-time aggregation	`GLOBAL WINDOW TRIGGER WHEN`	`FIRE_AND_PURGE`, aggregate-value / compound-condition triggers

How to run these SQL queries

Each scenario below shows only SQL + input + output; for how to run them see Case Studies · How to Run Case SQL. Non-aggregation queries (scenario A) use EmitSync for synchronous single-row results; aggregation/window queries (scenarios B–E) use Emit + AddSink for batch callbacks.

# Scenario A: Single-Point Temperature Alert (Non-Aggregation, EmitSync)

# Business Goal

As each temperature reading arrives, immediately do the Fahrenheit conversion, grade it (OK/WARNING/CRITICAL), and dispatch only readings above 30. The call must be one in / one out, synchronous and blocking, with the caller getting the result directly.

# SQL

SELECT deviceId,
       temperature,
       temperature * 1.8 + 32 AS temp_f,
       CASE WHEN temperature > 35 THEN 'CRITICAL'
            WHEN temperature > 30 THEN 'WARNING'
            ELSE 'OK' END AS level
FROM stream
WHERE temperature > 30

1
2
3
4
5
6
7
8

# Input and Output

Input 4 readings (28 fails temperature > 30 and is filtered):

{"deviceId": "dev-01", "temperature": 28.0}
{"deviceId": "dev-02", "temperature": 32.0}
{"deviceId": "dev-03", "temperature": 38.0}
{"deviceId": "dev-04", "temperature": 42.0}

1
2
3
4

Use EmitSync to get the result synchronously per row; output (28 filtered):

[filtered] dev-01
[alert]   {deviceId:dev-02 temperature:32 temp_f:89.6  level:WARNING}
[alert]   {deviceId:dev-03 temperature:38 temp_f:100.4 level:CRITICAL}
[alert]   {deviceId:dev-04 temperature:42 temp_f:107.6 level:CRITICAL}

1
2
3
4

# Behavior Notes

EmitSync synchronously returns a single-row result for non-aggregation queries; a row that misses WHERE returns nil (not an error), which lets you distinguish "filtered" from "error".
Calling EmitSync on an aggregation query errors: synchronous mode only supports non-aggregation queries.
CASE WHEN ... THEN ... ELSE ... END works both for grading columns and, combined with WHERE, for conditional filtering.

# Scenario B: Rolling 1-Minute Device Metrics (Event Time + Watermark)

# Business Goal

Per device, emit once per 1-minute window: sample count, average/max/min temperature. Requirements:

Use the timestamp carried by the reading (event time), not processing time;
Tolerate 5 seconds of out-of-order arrival caused by network jitter;
Still close the window when a device goes offline — no dead data left behind.

# SQL

SELECT deviceId,
       COUNT(*) AS samples,
       AVG(temperature) AS avg_t,
       MAX(temperature) AS max_t,
       MIN(temperature) AS min_t,
       window_start()  AS w_start
FROM stream
WHERE temperature IS NOT NULL
GROUP BY deviceId, TumblingWindow('1m')
WITH (TIMESTAMP='ts', TIMEUNIT='ms', MAXOUTOFORDERNESS='5s', IDLETIMEOUT='10s')

1
2
3
4
5
6
7
8
9
10

Meaning of each WITH option:

Option	Meaning
`TIMESTAMP='ts'`	Use field `ts` as the event time
`TIMEUNIT='ms'`	`ts` is in milliseconds (default `s`)
`MAXOUTOFORDERNESS='5s'`	Watermark = `max(event_time) - 5s`; tolerate 5s of out-of-order
`IDLETIMEOUT='10s'`	After the source is idle for 10s, advance the watermark by processing time so an offline device does not stall the window

# Input and Output

The first 5 readings all fall in the first 1-minute window [0, 60000ms), some intentionally out of order (within the 5s tolerance); the final ts=66000 pushes the watermark past 60000 and fires the first window:

{"deviceId": "dev-01", "ts": 1000,  "temperature": 31.0}
{"deviceId": "dev-01", "ts": 3000,  "temperature": 33.0}
{"deviceId": "dev-02", "ts": 2000,  "temperature": 29.0}
{"deviceId": "dev-01", "ts": 1500,  "temperature": 30.0}
{"deviceId": "dev-02", "ts": 4000,  "temperature": 27.0}
{"deviceId": "dev-03", "ts": 66000, "temperature": 25.0}

1
2
3
4
5
6

Output (batch callback on window close, one row per device):

[window] {deviceId:dev-01 samples:3 avg_t:31.33 max_t:33 min_t:30 w_start:0}
[window] {deviceId:dev-02 samples:2 avg_t:28    max_t:29 min_t:27 w_start:0}

1
2

# Behavior Notes

Output is not per-row; it is a batch callback on window close, one row per deviceId within the window.
dev-01@1500 arrives after dev-01@3000 but is still placed in the window correctly thanks to MAXOUTOFORDERNESS='5s'.
If no further data arrives, IDLETIMEOUT='10s' advances the watermark by processing time after 10s of silence and closes the window.
window_start() returns a nanosecond timestamp; divide by 1e6 to display in milliseconds.

# Scenario C: High-Cardinality Device Anomaly Detection (CountingWindow + State TTL)

# Business Goal

The fleet is large (tens of thousands to hundreds of thousands of devices), and a fraction of them report a few times and go offline. Requirement: every 50 readings, emit that device's stats (mean + standard deviation), while not letting "never-quite-reaches-50" devices blow up memory.

# SQL

SELECT deviceId,
       COUNT(*)           AS cnt,
       AVG(temperature)   AS avg_t,
       STDDEV(temperature) AS stddev_t
FROM stream
GROUP BY deviceId, CountingWindow(50)
WITH (STATETTL='1h')

1
2
3
4
5
6
7

Key points:

CountingWindow(50) triggers by count, independent of time, and supports processing time only (it ignores TIMESTAMP/TIMEUNIT).
STATETTL='1h': group keys that have not triggered again for over 1 hour are reaped by a background goroutine (lazy, in the Start goroutine).
The default STATETTL=0 means disabled (Flink-aligned). That is, without STATETTL, a device that reports only a few times leaves its counter state in memory forever — the "high-cardinality dead-key" problem.

# Input and Output

dev-A reports 50 times consecutively (temperature cycling regularly between 20.0 and 24.0) and fires when full; dev-B reports only 3 times and vanishes, never fires, and is reaped by STATETTL:

{"deviceId": "dev-A", "temperature": 20.0}   // ×50, temperature 20.0..24.0
{"deviceId": "dev-B", "temperature": 99.0}   // ×3, then disappears

1
2

Output (only the filled dev-A):

[burst] {deviceId:dev-A cnt:50 avg_t:22 stddev_t:1.414}

# Behavior Notes

dev-A fires the moment it hits 50: cnt=50, avg_t=22.0 (mean of 20..24), stddev_t≈1.414 (population standard deviation).
dev-B never appears in the output, but its state would otherwise be retained: without STATETTL, many such devices inflate memory; with STATETTL='1h', any key inactive for over 1 hour is reaped.
STDDEV is population standard deviation; use STDDEVS for sample standard deviation.

# Scenario D (Bonus): Deterministic Tests with TriggerWindow

# Business Goal

Time windows are awkward in tests — you do not want to literally time.Sleep(5*time.Second) waiting for a window to close. TriggerWindow() lets you manually trigger the current window to emit immediately, making tests both fast and deterministic.

# SQL

SELECT deviceId, COUNT(*) AS cnt
FROM stream
GROUP BY deviceId, TumblingWindow('5s')

1
2
3

# Input and Output

Emit just one row, then call ssql.TriggerWindow() to flush the window immediately, without waiting 5 seconds:

{"deviceId": "dev-X"}

[triggered] {deviceId:dev-X cnt:1}

# Behavior Notes

TriggerWindow() immediately flushes the current time window; for CountingWindow it is a no-op (count-driven, time-independent); for non-window queries it is also a safe no-op.
So it is safe to put in a test teardown: regardless of the SQL shape under test, it will never panic.

# Scenario E: Global-Window Condition-Driven Alert (GLOBAL WINDOW + TRIGGER WHEN)

# Business Goal

Scenario A judges a single-point temperature the moment each reading arrives; but some alerts depend on a cumulative state — "alert only when the running max temperature for that device crosses the threshold", or "settle once every N readings". Such "emit only when a condition hits" needs have no fixed time boundary and not necessarily a count, which corresponds to StreamSQL's global window (aligned with Flink GlobalWindows + Trigger): no built-in boundary, never auto-fires by default, driven by a TRIGGER WHEN predicate — each datum updates the running aggregate, and when the predicate hits it emits the current result and clears that group's state (FIRE_AND_PURGE).

# SQL

SELECT deviceId,
       MAX(temperature) AS max_t,
       COUNT(*)         AS samples
FROM stream
GROUP BY deviceId, GLOBAL WINDOW TRIGGER WHEN MAX(temperature) > 50

1
2
3
4
5

# Input and Output

dev-01 heats up step by step; 55 crosses the threshold and fires immediately (max_t=55, samples=3) then resets; the subsequent 48 enters a new group and does not cross:

{"deviceId": "dev-01", "temperature": 40.0}
{"deviceId": "dev-01", "temperature": 45.0}
{"deviceId": "dev-01", "temperature": 55.0}
{"deviceId": "dev-01", "temperature": 48.0}

1
2
3
4

[global] {deviceId:dev-01 max_t:55 samples:3}

# Behavior Notes

When 40 and 45 arrive, the running max=45 ≤ 50 does not fire; when 55 arrives, max=55 > 50 hits the predicate, emits the group's current aggregate (max_t=55, samples=3), and clears that group's state.
The subsequent 48 enters a brand-new group and reaccumulates from 0; max=48 ≤ 50 does not fire again. If a 52 then arrived, it would fire again immediately (max_t=52, samples=1) — this is the FIRE_AND_PURGE "fire-and-reset, next batch starts over" semantics.
Key difference from Scenario A: Scenario A judges a single-point value per row (temperature > 30); here what is judged is the cumulative max (MAX(temperature) > 50), which only a global window can do.

How to Run Case SQL — the unified Go boilerplate
Change Data Capture — analytic functions for change detection
SQL Reference — full SQL syntax and function list
API Reference — EmitSync / Emit / AddSink / TriggerWindow in detail

Edit this page on GitHub

Last Updated: 2026/07/11, 04:40:00

← Data Filtering and Transformation

IoT Temperature Alerting and Metrics Aggregation

# IoT Temperature Alerting and Metrics Aggregation

# Case Overview

# Scenario A: Single-Point Temperature Alert (Non-Aggregation, EmitSync)

# Business Goal

# SQL

# Input and Output

# Behavior Notes

# Scenario B: Rolling 1-Minute Device Metrics (Event Time + Watermark)

# Business Goal

# SQL

# Input and Output

# Behavior Notes

# Scenario C: High-Cardinality Device Anomaly Detection (CountingWindow + State TTL)

# Business Goal

# SQL

# Input and Output

# Behavior Notes

# Scenario D (Bonus): Deterministic Tests with TriggerWindow

# Business Goal

# SQL

# Input and Output

# Behavior Notes

# Scenario E: Global-Window Condition-Driven Alert (GLOBAL WINDOW + TRIGGER WHEN)

# Business Goal

# SQL

# Input and Output

# Behavior Notes

# 📚 Related Docs