流式计算

# 流式计算组件

RuleGo 提供了基于 StreamSQL 引擎的流式计算组件，支持使用SQL语法进行实时数据处理。

# 组件列表

# streamTransform

节点类型： x/streamTransform

流转换器组件，逐条同步处理非聚合 SQL 查询（走直连路径，状态跨事件保留），支持：

数据过滤、字段选择、重命名和计算
条件过滤和数据验证
单条和批量数据处理
分析函数：lag/latest、had_changed/changed_col/changed_cols、acc_*（变化检测、生命周期累计）
流-表 JOIN：用元数据表富化流行
60+ 内置函数

适用场景： 实时数据清洗、格式转换、变化检测（CDC）、生命周期累计

# streamAggregator

节点类型： x/streamAggregator

流聚合器组件，处理带窗口/分组的聚合 SQL 查询（异步触发，聚合结果走 window_event 链），支持：

窗口聚合（滚动、滑动、计数、会话、全局窗口）
分组聚合和多维度统计
聚合函数（COUNT、SUM、AVG、MAX、MIN 等）
HAVING 过滤聚合结果；流-表 JOIN 富化后再聚合
窗口内可用分析函数对窗口输出做变化检测/回溯/累计

适用场景： 实时统计分析、监控报警、持续超阈值检测、数据汇总

如何选择

逐条、立刻出结果（过滤/转换/变化检测/累计）→ streamTransform
攒一批、窗口触发再出结果（统计/持续检测）→ streamAggregator 分析函数不带窗口时进 transform，带窗口（对窗口输出求值）时进 aggregator。

# 快速开始

# 1. 安装依赖

go get github.com/rulego/rulego-components

# 2. 注册组件

import _ "github.com/rulego/rulego-components/stats/streamsql"

1
2

# 3. 使用示例

# 数据转换示例

{
  "id": "transform1",
  "type": "x/streamTransform",
  "name": "温度转换",
  "configuration": {
    "sql": "SELECT deviceId, temperature, temperature * 1.8 + 32 as temp_fahrenheit FROM stream WHERE temperature > 0"
  }
}

1
2
3
4
5
6
7
8

# 数据聚合示例

{
  "id": "aggregator1",
  "type": "x/streamAggregator",
  "name": "温度统计",
  "configuration": {
    "sql": "SELECT deviceId, AVG(temperature) as avg_temp, COUNT(*) as count FROM stream GROUP BY deviceId, TumblingWindow('5m')"
  }
}

1
2
3
4
5
6
7
8

# 变化检测示例（分析函数）

{
  "id": "cdc1",
  "type": "x/streamTransform",
  "name": "电流突变检测",
  "configuration": {
    "sql": "SELECT current, deviceId FROM stream WHERE current > 300 AND lag(current) OVER (PARTITION BY deviceId) < 300"
  }
}

1
2
3
4
5
6
7
8

# 应用场景

# IoT数据处理

传感器数据清洗和格式化
实时温度、湿度等指标监控
设备状态统计和报警

# 实时监控

系统性能指标聚合
异常检测和告警
实时仪表板数据处理

# 数据分析

流式数据预处理
实时统计计算
多维度数据分析

在 GitHub 上编辑此页

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

← 串口通信流聚合器→