流聚合器

# streamAggregator

节点类型： x/streamAggregator

说明： 流聚合器节点，用于处理聚合SQL查询，如窗口聚合、分组聚合等。该组件基于 StreamSQL 引擎，支持滚动窗口、滑动窗口等多种窗口类型的聚合计算。支持单条数据和数组数据输入。

# 输入数据支持

该节点支持两种输入数据格式：

# 单条数据输入

直接处理单个JSON对象：

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

# 数组数据输入

自动处理JSON数组，将数组中的每个元素逐条添加到聚合流中：

[
  {"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

数组处理说明

数组中的每个元素都会被逐条添加到聚合流中参与聚合计算
原始数组消息会通过Success链继续传递，保持数据流的连续性
聚合结果仍然通过window_event链传递

# 配置

字段	类型	说明	默认值
sql	string	聚合 SQL 查询语句，必须包含聚合函数或窗口函数	无
tables	array	可选，元数据表配置（流-表 JOIN 富化后再聚合，见SQL 参考）	无

# SQL语法支持

详细语法参考

完整的 SQL 语法说明请参考：StreamSQL SQL语法参考

# 关系类型

Success： 原始消息成功处理后，通过此关系链传递原始消息
window_event： 聚合结果通过此关系链传递，消息体为聚合计算的结果，结果格式是一个多列数组
Failure： 处理失败时，通过此关系链传递错误信息

# 执行结果

# Success链输出

原始消息不变，继续传递给下一个节点。

# window_event链输出

聚合结果作为新消息传递，消息格式：

[
  {
    "field1": "value1",
    "field2": "value2",
    "count": 10,
    "avg_temperature": 25.5
  }
]

1
2
3
4
5
6
7
8

# Failure链输出

错误信息，包含具体的错误描述。

# 配置示例

# 基础分组聚合

{
  "id": "s1",
  "type": "x/streamAggregator",
  "name": "设备温度聚合",
  "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

# 滑动窗口聚合

{
  "id": "s2",
  "type": "x/streamAggregator",
  "name": "滑动窗口分析",
  "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

# 多字段聚合

{
  "id": "s3",
  "type": "x/streamAggregator",
  "name": "多维度聚合",
  "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

# 窗口内分析函数

分析函数（lag、had_changed、changed_col/changed_cols、acc_*）可用在聚合 SQL 的 SELECT 中，对窗口产出行求值，状态跨窗口保留（不随窗口清零）。参数须是聚合函数或 GROUP BY 字段，不能引用裸原始列。

-- 每两个事件求一次均值，只在均值变化时输出该窗口
SELECT changed_cols("t", true, avg(temperature)) FROM stream GROUP BY CountingWindow(2)

-- 跨窗口累计窗口均值
SELECT acc_sum(avg(temperature)) AS total FROM stream GROUP BY CountingWindow(2)

-- 先 JOIN 元数据富化，再按位置分组聚合
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

完整语法见分析函数。

# 应用示例

# 示例1：设备状态监控

场景： 监控IoT设备的温度数据，每2秒计算一次各设备的平均温度和最大温度。

规则链配置：

{
  "ruleChain": {
    "id": "device_monitoring",
    "name": "设备监控规则链",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamAggregator",
        "name": "温度聚合",
        "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": "结果处理",
        "configuration": {
          "jsScript": "msg.timestamp = new Date().toISOString(); return {'msg': msg, 'metadata': metadata, 'msgType': msgType};"
        }
      },
      {
        "id": "s3",
        "type": "log",
        "name": "聚合结果日志",
        "configuration": {
          "jsScript": "return 'Aggregation Result: ' + JSON.stringify(msg);"
        }
      },
      {
        "id": "s4",
        "type": "log",
        "name": "原始数据日志",
        "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
49
50
51
52
53
54
55
56
57
58
59
60

输入数据：

{"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

聚合结果输出：

{
  "deviceId": "device001",
  "avg_temp": 25.85,
  "max_temp": 26.2,
  "count": 2
}

1
2
3
4
5
6

# 示例2：高温报警系统

场景： 使用滑动窗口监控温度变化，当3秒内平均温度超过30度时触发报警。

规则链配置：

{
  "ruleChain": {
    "id": "temperature_alarm",
    "name": "高温报警规则链",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamAggregator",
        "name": "温度滑动窗口",
        "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": "高温过滤",
        "configuration": {
          "jsScript": "return msg.avg_temp > 30;"
        }
      },
      {
        "id": "s3",
        "type": "jsTransform",
        "name": "报警信息",
        "configuration": {
          "jsScript": "msg.alert = 'High temperature detected!'; msg.level = 'WARNING'; return {'msg': msg, 'metadata': metadata, 'msgType': 'ALARM'};"
        }
      },
      {
        "id": "s4",
        "type": "log",
        "name": "报警日志",
        "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
49
50
51
52
53
54
55
56
57
58
59
60

# 示例3：数组数据批量聚合

场景： 处理包含多个设备数据的数组消息，进行批量聚合分析。

输入数据：

[
  {"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

规则链配置：

{
  "ruleChain": {
    "id": "batch_aggregation",
    "name": "批量数据聚合",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamAggregator",
        "name": "按位置聚合",
        "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": "聚合结果",
        "configuration": {
          "jsScript": "return 'Location Aggregation: ' + JSON.stringify(msg);"
        }
      },
      {
        "id": "s3",
        "type": "log",
        "name": "原始数组",
        "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
40
41
42
43
44
45
46
47

聚合结果输出：

[
  {"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

# 示例4：持续超阈值告警（滑动窗口 + HAVING）

场景： 电流持续 10 秒都大于 200A 才告警——中间出现过下跌则不告警。

节点配置：

{
  "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

原理： 滑动窗口聚合这 10 秒内的全部事件（含 <200 的下跌），HAVING mn > 200 把出现过下跌的窗口过滤掉——留下来的就是"全程 >200"。没有任何窗口满足 HAVING 时，该周期不输出。

HAVING 引用别名

HAVING 引用 SELECT 里的别名（mn），不能复述聚合函数（HAVING min(current) > 200 不生效）。

# 示例5：窗口内变化检测（分析函数）

场景： 多设备同流，每两个采样求一次平均，只在某设备窗口均值发生变化时输出。

节点配置：

{
  "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

输入/输出（A: 10,20,30,40；B: 5,5）：

A 窗口均值 15（首→变化）、35（变化）  → {deviceId:"A", chg:15} {deviceId:"A", chg:35}
B 窗口均值 5（首→变化）              → {deviceId:"B", chg:5}

1
2

分析函数在窗口产出后对均值求值，状态跨窗口保留；默认按 GROUP BY 键分区，各设备互不串扰。

# 示例6：元数据富化后聚合（流-表 JOIN）

场景： 设备上报只有 deviceId，先用元数据表富化出 location，再按位置聚合平均温度。

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

节点配置：

{
  "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

输出：

[
  {"location": "厂房A", "avg_temp": 25.5, "cnt": 3},
  {"location": "厂房B", "avg_temp": 22.1, "cnt": 2}
]

1
2
3
4

# 注意事项

SQL语法限制： 只支持聚合查询，不支持非聚合的SELECT语句
窗口类型： 必须在GROUP BY子句中指定窗口函数
性能考虑： 窗口大小和滑动间隔会影响内存使用和计算性能
数据类型： 确保聚合字段的数据类型支持相应的聚合函数
数组处理： 数组中的每个元素都会逐条添加到聚合流中，原始数组消息通过Success链传递
窗口事件回调： 窗口事件触发的结束回调需要通过 Config.OnEnd 设置，而不是通过 OnMsg 注册的 OnEnd 回调。这是因为窗口事件是由聚合器内部触发的，不会经过常规的消息处理流程

# 窗口事件回调示例

正确的方式 - 使用 Config.OnEnd：

// 设置全局聚合结果处理器
config.OnEnd = func(ctx types.RuleContext, msg types.RuleMsg, err error, relationType string) {
    if err == nil && msg.Type == WindowEventMsgType {
        // 处理窗口聚合结果
        var result map[string]interface{}
        if jsonErr := json.Unmarshal([]byte(msg.Data.String()), &result); jsonErr == nil {
            // 处理聚合结果
            fmt.Printf("聚合结果: %+v\n", result)
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11

错误的方式 - 使用 OnMsg 的 OnEnd：

// 这种方式无法捕获窗口事件
ruleEngine.OnMsg(msg, types.WithOnEnd(func(ctx types.RuleContext, msg types.RuleMsg, err error, relationType string) {
    // 窗口事件不会触发这个回调
}))

1
2
3
4

在 GitHub 上编辑此页

上次更新: 2026/07/11, 11:54:34

← 流式计算流转换器→