Stream Processing

# Stream Processing Components

RuleGo provides stream processing components based on the StreamSQL engine, supporting real-time data processing using SQL syntax.

# Component List

# streamTransform

Node Type: x/streamTransform

Stream Transformer component, processes non-aggregate SQL queries synchronously per event (direct path, state preserved across events), supports:

Data filtering, field selection, renaming, and calculation
Conditional filtering and data validation
Single and batch data processing
Analytic functions: lag/latest, had_changed/changed_col/changed_cols, acc_* (change detection, lifetime accumulation)
Stream-table JOIN: enrich stream rows with metadata tables
60+ built-in functions

Applicable Scenarios: Real-time data cleaning, format conversion, change detection (CDC), lifetime accumulation

# streamAggregator

Node Type: x/streamAggregator

Stream Aggregator component, processes aggregate SQL queries with windows/grouping (asynchronously triggered, results go through the window_event chain), supports:

Window aggregation (Tumbling, Sliding, Counting, Session, Global windows)
Group aggregation and multi-dimensional statistics
Aggregation functions (COUNT, SUM, AVG, MAX, MIN, etc.)
HAVING to filter aggregation results; stream-table JOIN to enrich before aggregating
Analytic functions usable inside windows for change detection/lookback/accumulation on window output

Applicable Scenarios: Real-time statistical analysis, monitoring alarms, persistent over-threshold detection, data summarization

How to choose

Per event, immediate result (filtering/transform/change detection/accumulation) → streamTransform
Batch up, emit on window trigger (statistics/persistent detection) → streamAggregator Analytic functions without a window go to transform; with a window (evaluated on window output) they go to aggregator.

# Quick Start

# 1. Install Dependencies

go get github.com/rulego/rulego-components

# 2. Register Components

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

1
2

# 3. Usage Examples

# Data Transformation Example

{
  "id": "transform1",
  "type": "x/streamTransform",
  "name": "Temperature Conversion",
  "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

# Data Aggregation Example

{
  "id": "aggregator1",
  "type": "x/streamAggregator",
  "name": "Temperature Statistics",
  "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

# Change Detection Example (Analytic Function)

{
  "id": "cdc1",
  "type": "x/streamTransform",
  "name": "Current Spike Detection",
  "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

# Application Scenarios

# IoT Data Processing

Sensor data cleaning and formatting
Real-time monitoring of indicators such as temperature and humidity
Device status statistics and alarms

# Real-time Monitoring

System performance indicator aggregation
Anomaly detection and alerting
Real-time dashboard data processing

# Data Analysis

Stream data preprocessing
Real-time statistical calculation
Multi-dimensional data analysis

Edit this page on GitHub

Last Updated: 2026/07/11, 11:54:34

← Serial Communication Stream Aggregator→