Aggregate Stream Node
Use this node when you need to compute rolling statistics over a stream of incoming messages — for example, hourly average temperature, daily total energy consumption, or the maximum pressure reading per shift. It groups messages by originator and aggregation interval, computes the result in memory, and emits a POST_TELEMETRY_REQUEST at the end of each interval.
Intervals can be persisted to storage periodically to avoid data loss on restart, and a default-value mechanism lets you fill in gaps for entities that stopped reporting.
Configuration
Section titled “Configuration”Aggregation
Section titled “Aggregation”| Field | Required | Description |
|---|---|---|
| Aggregation function | Yes | MIN, MAX, SUM, AVG (default), COUNT, or COUNT_UNIQUE |
| Input value key | Yes | The incoming telemetry key to aggregate. Default: temperature |
| Output value key | Yes | The key of the aggregated value in the output message. Default: avgHourlyTemperature |
| Interval | Yes | The time window over which values are aggregated: a calendar interval (hour, day, week, month, year) or a custom one. Default: 1 hour |
| Output message type | Yes | The type of the output message. Default: POST_TELEMETRY_REQUEST |
Persistence
Section titled “Persistence”| Field | Description |
|---|---|
| Interval persistence policy | When the node emits interval results: On each check, On each check after interval end (default), or On each message |
| Interval check value | How often the node checks the intervals. Default: 1 minute |
| State persistence policy | How often the aggregation state is saved: On each change (default) or Periodically |
| State persistence value | Interval between state saves, for the Periodically policy. Default: 1 minute |
Default values
Section titled “Default values”| Field | Description |
|---|---|
| Create intervals automatically | When enabled, default-value messages are emitted for entities in the configured entity group even if no data arrives during an interval |
| Interval entities | Entity group whose members receive default-value messages when no data is reported |
Output message
Section titled “Output message”The node emits a message of the configured Output message type (POST_TELEMETRY_REQUEST by default). The body holds the aggregated value under the Output value key:
{ "<output value key>": <aggregated value>}The metadata holds ts, the timestamp of the interval, and overwriteValue = true.
The originator of the output message is the same as the originator of the input messages.
Output connections
Section titled “Output connections”| Connection | Condition |
|---|---|
Success |
Aggregated result message, emitted according to the interval persistence policy |
Failure |
Unexpected error during processing |
Example — Hourly average temperature
Section titled “Example — Hourly average temperature”Configuration: Function = AVG, Input value key = temperature, Output value key = avgHourlyTemperature, Interval = 1 hour, Interval persistence = On each check after interval end.
Incoming messages: a stream of POST_TELEMETRY_REQUEST messages from a temperature sensor with key temperature.
Result: at the end of each hour, one message is emitted with the average temperature for that hour:
{ "avgHourlyTemperature": 22.3}with metadata ts = 1700006400000.
This message can be piped directly into a Save time series node to persist the hourly average alongside raw readings.
Was this helpful?