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 (or a custom message type) according to the configured interval persistence policy. The running aggregator can be persisted to storage so values are not lost on restart, and a default‑value mechanism lets you fill in gaps for entities that stopped reporting.

This node is order-sensitive: delayed or out-of-order messages that arrive after their interval has already ended are silently ignored and not included in the aggregated result. Duplicate messages are also counted.

If your data sources produce late-arriving telemetry or you need the ability to re-process historical data, consider using Time Series Data Aggregation or Related Entities Aggregation (Calculated Fields) instead — they aggregate from stored data on demand and are not affected by message ordering or duplicates.

A few terms recur throughout the configuration; defining them up front makes the rest of the page short.

Interval. A bucket keyed by (originatorId, intervalStartTs). For each originator the node maintains an in‑memory map of intervals — one per time window (hour, day, custom, …). Each bucket holds a small aggregator object (e.g. {sum, count} for Average, the smallest value seen for Min, a set of values for Count unique). An interval is created lazily, the first time a message for that originator falls into the window. Old intervals are evicted from memory after a TTL of 2 × the interval duration.

State. The aggregator object inside an interval. It is the small “scratch‑pad” the node uses to accumulate the result — not the user’s telemetry. Every incoming message updates the matching state.

Reporting. Emitting an outgoing message (default POST_TELEMETRY_REQUEST) with the current aggregated value on the Success branch. Reporting does not write to the database by itself — to store the result you must wire Success to a Save Time Series node.

State persistence. Writing the in‑memory state back to the time‑series DB so the node can resume aggregation after a restart, partition switch, or rule‑chain reload. The state is stored as a JSON string under a special key prefix (ruleNodeState_<ruleNodeId>) and uses saveWithoutLatest, so it never appears in the entity’s latest telemetry. This is independent of, and additional to, any user‑level Save Time Series step you might do downstream.

Check. A periodic internal self‑tick re‑scheduled every Interval check value units (minimum 1 minute). At each check the node walks the interval table, decides which intervals to report, and evicts intervals past TTL. A separate internal tick drives state persistence when the Periodically policy is selected.

The fields below follow the order of the rule node editor.

Controls when an aggregate result is emitted downstream. Despite the field name, this setting does not by itself write to the database — it only triggers an outgoing message on the Success branch.

Controls when the in‑memory aggregator is written to the time‑series DB so values are not lost on restart. Each write goes to a special internal key prefix (ruleNodeState_<ruleNodeId>) and never appears in the entity’s latest telemetry.

By default an interval is created only when a message for that originator arrives. If a device goes silent, no interval and no result message are produced for the missing windows. Tick this checkbox to emit results for silent entities.

When enabled the node periodically iterates the configured entity set and fills any missing intervals between the last seen one and “now” with a default‑valued state. Those default intervals follow the normal report path, so a result is emitted even when no telemetry arrived.

Ticking the checkbox reveals an Interval entities card with the following structure.

A nested card under Interval entities. Selects the universe of entities for which default‑value intervals are generated. Pick one mode via the radio button.

Optional. Selects a rule engine queue for tick and outgoing messages. If empty, the node uses the queue of the rule chain.

At each emission the node produces a message of the configured Output message type (default POST_TELEMETRY_REQUEST). The originator is the entity whose interval is being reported.

{
  "<output value key>": <aggregated value>
}

Metadata: ts is set to the interval start timestamp, and overwriteValue=true so a downstream Save Time Series node replaces any earlier write at the same timestamp.

Three sources of writes to consider when sizing the DB:

1. Reporting never writes to the DB by itself. If you wire Success to a Save Time Series node you get one telemetry write per emission, per originator, per output key.
2. State persistence:
   • On each change ≈ 1 internal write per accepted incoming message.
   • Periodically ≈ 1 internal write per dirty (entity, interval) per state‑persist tick (batched).
3. Create intervals automatically multiplies emissions by the size of the configured entity set, regardless of how many of those entities actually sent data.

Configuration: Aggregation function = Average, Aggregate interval type = Hour, Interval persistence policy = On each check after interval end, State persistence policy = On each change.

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
}

This message can be piped directly into a Save time series node to persist the hourly average alongside raw readings.