RuleGo
目录

Stream Aggregator

# streamAggregator

Node Type: x/streamAggregator

Description: Stream aggregator node, used for processing aggregate SQL queries, such as window aggregation, group aggregation, etc. This component is based on the StreamSQL engine and supports aggregation calculations for various window types such as Tumbling Window and Sliding Window. Supports single data and array data input.

# Input Data Support

This node supports two input data formats:

# Single Data Input

Directly process a single JSON object:

{"deviceId": "sensor001", "temperature": 25.5, "humidity": 60.2}
1

# Array Data Input

Automatically process JSON arrays, adding each element in the array to the aggregation stream one by one:

[
  {"deviceId": "sensor001", "temperature": 25.5, "humidity": 60.2},
  {"deviceId": "sensor002", "temperature": 28.3, "humidity": 55.8},
  {"deviceId": "sensor003", "temperature": 22.1, "humidity": 65.4}
]
1
2
3
4
5

Array Processing Description

  • Each element in the array will be added to the aggregation stream one by one to participate in the aggregation calculation.
  • The original array message will continue to be passed through the Success chain, maintaining data flow continuity.
  • Aggregation results are still passed through the window_event chain.

# Configuration

Field Type Description Default Value
sql string Aggregate SQL query statement, must contain aggregation functions or window functions None
tables array Optional, metadata table configuration for stream-table JOIN enrichment (see SQL Reference) None

# SQL Syntax Support

Detailed Syntax Reference

For complete SQL syntax instructions, please refer to: StreamSQL SQL Syntax Reference

# Relation Types

  • Success: After the original message is successfully processed, the original message is passed through this relation chain.
  • window_event: Aggregation results are passed through this relation chain. The message body is the result of the aggregation calculation, and the result format is a multi-column array.
  • Failure: When processing fails, error information is passed through this relation chain.

# Execution Results

# Success Chain Output

The original message remains unchanged and continues to be passed to the next node.

# window_event Chain Output

The aggregation result is passed as a new message, format:

[
  {
    "field1": "value1",
    "field2": "value2",
    "count": 10,
    "avg_temperature": 25.5
  }
]
1
2
3
4
5
6
7
8

# Failure Chain Output

Error message, containing specific error descriptions.

# Configuration Examples

# Basic Group Aggregation

{
  "id": "s1",
  "type": "x/streamAggregator",
  "name": "Device Temperature Aggregation",
  "configuration": {
    "sql": "SELECT deviceId, AVG(temperature) as avg_temp, MAX(temperature) as max_temp, COUNT(*) as count FROM stream GROUP BY deviceId, TumblingWindow('2s')"
  }
}
1
2
3
4
5
6
7
8

# Sliding Window Aggregation

{
  "id": "s2",
  "type": "x/streamAggregator",
  "name": "Sliding Window Analysis",
  "configuration": {
    "sql": "SELECT AVG(temperature) as avg_temp, COUNT(*) as count FROM stream GROUP BY SlidingWindow('10s', '2s')"
  }
}
1
2
3
4
5
6
7
8

# Multi-field Aggregation

{
  "id": "s3",
  "type": "x/streamAggregator",
  "name": "Multi-dimensional Aggregation",
  "configuration": {
    "sql": "SELECT deviceType, location, AVG(temperature) as avg_temp, MIN(humidity) as min_humidity, MAX(pressure) as max_pressure FROM stream GROUP BY deviceType, location, TumblingWindow('5m')"
  }
}
1
2
3
4
5
6
7
8

# Windowed Analytic Functions

Analytic functions (lag, had_changed, changed_col/changed_cols, acc_*) can be used in the SELECT of an aggregate SQL query. They are evaluated against the window output rows, and their state is preserved across windows (not cleared on window close). Their arguments must be aggregate functions or GROUP BY fields — they cannot reference raw columns.

-- Average every two events, output the window only when the average changes
SELECT changed_cols("t", true, avg(temperature)) FROM stream GROUP BY CountingWindow(2)

-- Accumulate window averages across windows
SELECT acc_sum(avg(temperature)) AS total FROM stream GROUP BY CountingWindow(2)

-- JOIN metadata to enrich, then group and aggregate by location
SELECT m.location, AVG(temperature) AS avg_temp FROM stream JOIN meta m ON deviceId = m.deviceId
GROUP BY m.location, TumblingWindow('5s')
1
2
3
4
5
6
7
8
9

For full syntax see Analytic Functions.

# Application Examples

# Example 1: Device Status Monitoring

Scenario: Monitor IoT device temperature data, calculating the average and maximum temperature of each device every 2 seconds.

Rule Chain Configuration:

{
  "ruleChain": {
    "id": "device_monitoring",
    "name": "Device Monitoring Rule Chain",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamAggregator",
        "name": "Temperature Aggregation",
        "configuration": {
          "sql": "SELECT deviceId, AVG(temperature) as avg_temp, MAX(temperature) as max_temp, COUNT(*) as count FROM stream GROUP BY deviceId, TumblingWindow('2s')"
        }
      },
      {
        "id": "s2",
        "type": "jsTransform",
        "name": "Result Processing",
        "configuration": {
          "jsScript": "msg.timestamp = new Date().toISOString(); return {'msg': msg, 'metadata': metadata, 'msgType': msgType};"
        }
      },
      {
        "id": "s3",
        "type": "log",
        "name": "Aggregation Result Log",
        "configuration": {
          "jsScript": "return 'Aggregation Result: ' + JSON.stringify(msg);"
        }
      },
      {
        "id": "s4",
        "type": "log",
        "name": "Original Data Log",
        "configuration": {
          "jsScript": "return 'Original Data: ' + JSON.stringify(msg);"
        }
      }
    ],
    "connections": [
      {"fromId": "s1", "toId": "s2", "type": "window_event"},
      {"fromId": "s1", "toId": "s4", "type": "Success"},
      {"fromId": "s2", "toId": "s3", "type": "Success"}
    ]
  }
}
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

Input Data:

{"deviceId": "device001", "temperature": 25.5, "timestamp": "2023-09-13T10:00:00Z"}
{"deviceId": "device001", "temperature": 26.2, "timestamp": "2023-09-13T10:00:01Z"}
{"deviceId": "device002", "temperature": 24.8, "timestamp": "2023-09-13T10:00:01Z"}
1
2
3

Aggregation Result Output:

{
  "deviceId": "device001",
  "avg_temp": 25.85,
  "max_temp": 26.2,
  "count": 2
}
1
2
3
4
5
6

# Example 2: High Temperature Alarm System

Scenario: Use a sliding window to monitor temperature changes and trigger an alarm when the average temperature within 3 seconds exceeds 30 degrees.

Rule Chain Configuration:

{
  "ruleChain": {
    "id": "temperature_alarm",
    "name": "High Temperature Alarm Rule Chain",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamAggregator",
        "name": "Temperature Sliding Window",
        "configuration": {
          "sql": "SELECT AVG(temperature) as avg_temp, MAX(temperature) as max_temp, COUNT(*) as count FROM stream GROUP BY SlidingWindow('3s', '1s')"
        }
      },
      {
        "id": "s2",
        "type": "jsFilter",
        "name": "High Temperature Filter",
        "configuration": {
          "jsScript": "return msg.avg_temp > 30;"
        }
      },
      {
        "id": "s3",
        "type": "jsTransform",
        "name": "Alarm Message",
        "configuration": {
          "jsScript": "msg.alert = 'High temperature detected!'; msg.level = 'WARNING'; return {'msg': msg, 'metadata': metadata, 'msgType': 'ALARM'};"
        }
      },
      {
        "id": "s4",
        "type": "log",
        "name": "Alarm Log",
        "configuration": {
          "jsScript": "return 'ALARM: ' + JSON.stringify(msg);"
        }
      }
    ],
    "connections": [
      {"fromId": "s1", "toId": "s2", "type": "window_event"},
      {"fromId": "s2", "toId": "s3", "type": "True"},
      {"fromId": "s3", "toId": "s4", "type": "Success"}
    ]
  }
}
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

# Example 3: Batch Aggregation of Array Data

Scenario: Process an array message containing multiple devices' data for batch aggregation.

Input Data:

[
  {"deviceId": "sensor001", "temperature": 25.5, "location": "room1"},
  {"deviceId": "sensor002", "temperature": 28.3, "location": "room1"},
  {"deviceId": "sensor003", "temperature": 22.1, "location": "room2"},
  {"deviceId": "sensor004", "temperature": 30.8, "location": "room2"}
]
1
2
3
4
5
6

Rule Chain Configuration:

{
  "ruleChain": {
    "id": "batch_aggregation",
    "name": "Batch Data Aggregation",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamAggregator",
        "name": "Aggregate by Location",
        "configuration": {
          "sql": "SELECT location, AVG(temperature) as avg_temp, MAX(temperature) as max_temp, COUNT(*) as device_count FROM stream GROUP BY location, TumblingWindow('5s')"
        }
      },
      {
        "id": "s2",
        "type": "log",
        "name": "Aggregation Result",
        "configuration": {
          "jsScript": "return 'Location Aggregation: ' + JSON.stringify(msg);"
        }
      },
      {
        "id": "s3",
        "type": "log",
        "name": "Original Array",
        "configuration": {
          "jsScript": "return 'Original Array: ' + JSON.stringify(msg);"
        }
      }
    ],
    "connections": [
      {"fromId": "s1", "toId": "s2", "type": "window_event"},
      {"fromId": "s1", "toId": "s3", "type": "Success"}
    ]
  }
}
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

Aggregation Result Output:

[
  {"location": "room1", "avg_temp": 26.9, "max_temp": 28.3, "device_count": 2},
  {"location": "room2", "avg_temp": 26.45, "max_temp": 30.8, "device_count": 2}
]
1
2
3
4

# Example 4: Persistent Over-Threshold Alarm (Sliding Window + HAVING)

Scenario: Trigger an alarm only when the current stays above 200A for 10 seconds — any dip in between cancels the alarm.

Node Configuration:

{
  "id": "s4",
  "type": "x/streamAggregator",
  "configuration": {
    "sql": "SELECT min(current) AS mn, count(*) AS c FROM stream GROUP BY SlidingWindow('10s', '1s') HAVING mn > 200"
  }
}
1
2
3
4
5
6
7

How it works: The sliding window aggregates all events in those 10 seconds (including dips below 200), and HAVING mn > 200 filters out any window that contained a dip — the survivors are exactly the "always > 200" windows. When no window satisfies HAVING, that period produces no output.

HAVING references the alias

HAVING references the alias from SELECT (mn); you cannot restate the aggregate function (HAVING min(current) > 200 does not work).

# Example 5: Windowed Change Detection (Analytic Function)

Scenario: Multiple devices in one stream; average every two samples and output only when a device's window average changes.

Node Configuration:

{
  "id": "s5",
  "type": "x/streamAggregator",
  "configuration": {
    "sql": "SELECT deviceId, changed_col(true, avg(temp)) AS chg FROM stream GROUP BY deviceId, CountingWindow(2)"
  }
}
1
2
3
4
5
6
7

Input/Output (A: 10,20,30,40; B: 5,5):

A window averages 15 (first→change), 35 (change)  → {deviceId:"A", chg:15} {deviceId:"A", chg:35}
B window average 5 (first→change)                 → {deviceId:"B", chg:5}
1
2

The analytic function evaluates the average after the window emits; state is preserved across windows. It partitions by the GROUP BY key by default, so devices never cross-contaminate.

# Example 6: Enrich then Aggregate (Stream-Table JOIN)

Scenario: Devices report only deviceId; first enrich with location from a metadata table, then aggregate average temperature by location.

device_meta.json: [{"deviceId":"d1","location":"Plant A"},{"deviceId":"d2","location":"Plant B"}]

Node Configuration:

{
  "id": "s6",
  "type": "x/streamAggregator",
  "configuration": {
    "sql": "SELECT m.location, AVG(temperature) AS avg_temp, COUNT(*) AS cnt FROM stream JOIN meta m ON deviceId = m.deviceId GROUP BY m.location, TumblingWindow('5s')",
    "tables": [
      {"name": "meta", "source": "file", "path": "/etc/rulego/device_meta.json", "format": "json", "refresh": "30s"}
    ]
  }
}
1
2
3
4
5
6
7
8
9
10

Output:

[
  {"location": "Plant A", "avg_temp": 25.5, "cnt": 3},
  {"location": "Plant B", "avg_temp": 22.1, "cnt": 2}
]
1
2
3
4

# Notes

  1. SQL Syntax Restriction: Only aggregate queries are supported; non-aggregate SELECT statements are not allowed.
  2. Window Type: A window function must be specified in the GROUP BY clause.
  3. Performance: Window size and sliding interval affect memory usage and computation performance.
  4. Data Type: Ensure the data type of the aggregated field supports the corresponding aggregation function.
  5. Array Processing: Each element in the array is added to the aggregation stream one by one; the original array message is passed through the Success chain.
  6. Window Event Callback: The end callback triggered by window events must be set via Config.OnEnd, not via the OnEnd callback registered with OnMsg. This is because window events are triggered internally by the aggregator and do not go through the regular message processing flow.

# Window Event Callback Example

Correct way - use Config.OnEnd:

// Set the global aggregation result handler
config.OnEnd = func(ctx types.RuleContext, msg types.RuleMsg, err error, relationType string) {
    if err == nil && msg.Type == WindowEventMsgType {
        // Handle window aggregation result
        var result map[string]interface{}
        if jsonErr := json.Unmarshal([]byte(msg.Data.String()), &result); jsonErr == nil {
            // Process the aggregation result
            fmt.Printf("Aggregation result: %+v\n", result)
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11

Wrong way - use OnMsg's OnEnd:

// This approach cannot capture window events
ruleEngine.OnMsg(msg, types.WithOnEnd(func(ctx types.RuleContext, msg types.RuleMsg, err error, relationType string) {
    // Window events do not trigger this callback
}))
1
2
3
4
Edit this page on GitHub (opens new window)
Last Updated: 2026/07/11, 11:54:34
Stream Processing
Stream Transformer

← Stream Transformer→

Theme by Vdoing | Copyright © 2023-2026 RuleGo Team | Apache 2.0 License