RuleGo
目录

Stream Transformer

# streamTransform

Node Type: x/streamTransform

Description: Stream transformer node, based on the StreamSQL engine, uses SQL syntax to filter, transform, and process fields of real-time data streams. It processes non-aggregate queries synchronously per event (state is preserved across events), supporting data filtering, field transformation, format conversion, change detection, and lifetime accumulation. Supports single data and array data input.

# Features

  • SQL Syntax: Use standard SQL syntax for data transformation, low learning cost.
  • Real-time Processing: Synchronous processing of single data and array data.
  • Field Operations: Supports field selection, renaming, calculation, and conditional filtering.
  • Function Support: 60+ built-in functions, including math, string, time, etc.
  • Analytic Functions: Cross-event stateful computation — lag/latest, had_changed/changed_col/changed_cols, acc_* (see Analytic Functions).
  • Stream-Table JOIN: Enrich stream rows with metadata tables (see SQL Reference).
  • Conditional Filtering: Supports WHERE clause for data filtering.
  • Array Processing: Automatically processes array data, transforms each element and merges the results.

# Input Data Support

This node supports two input data formats:

# Single Data Input

Directly process a single JSON object: successful transformation goes to Success; WHERE not matched goes to Filtered; a processing error goes to Failure:

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

# Array Data Input

Automatically process JSON arrays, traverse each element for transformation, and merge the successfully transformed results into a new array for output:

[
  {"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 processed by SQL transformation one by one.
  • Only elements that are successfully transformed and meet the WHERE condition will be included in the output array.
  • If at least one element is successfully transformed, the merged array is output through the Success chain.
  • If all elements are filtered by WHERE (no errors), the merged result goes to Filtered; if any element errors, it goes to Failure.
  • Message metadata will contain processing statistics: originalCount, transformedCount, failedCount.

# Configuration

Field Type Description Default Value
sql string Transformation SQL statement, must be a non-aggregate query (cannot contain GROUP BY / aggregation functions; analytic functions are allowed) None
tables array Optional, metadata table configuration for stream-table JOIN enrichment (see below) None

# SQL Syntax Support

Detailed Syntax Reference

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

# Relation Types

  • Success: Successful transformation; the transformed data is passed through this relation chain (metadata match=true).
  • Filtered: Data was filtered (WHERE not matched, or changed_cols/changed_col had no change) — not an error; metadata match=false.
  • Failure: A processing error occurred (non-JSON data, SQL evaluation error, etc.); error information is passed through this relation chain.

Distinguishing "filtered" from "error"

Being filtered by WHERE or having no changed_cols change is an intentional no-output, routed to Filtered. Only genuine processing errors go to Failure. So the Failure chain can safely be wired to alerts/error handling without being polluted by normal event compression.

# Execution Results

# Success Chain Output

Transformed data, the format is determined by the SQL query result:

{
  "field1": "transformed_value1",
  "field2": "transformed_value2",
  "calculated_field": 123.45
}
1
2
3
4
5

# Failure Chain Output

Error message, containing specific error descriptions.

# Configuration Examples

# Basic Field Transformation

{
  "id": "s1",
  "type": "x/streamTransform",
  "name": "Temperature Unit Conversion",
  "configuration": {
    "sql": "SELECT deviceId, temperature, humidity, temperature * 1.8 + 32 as temp_fahrenheit FROM stream WHERE temperature IS NOT NULL",
    "debug": false
  }
}
1
2
3
4
5
6
7
8
9

# Data Filtering and Calculation

{
  "id": "s2",
  "type": "x/streamTransform",
  "name": "High Temperature Data Processing",
  "configuration": {
    "sql": "SELECT deviceId, temperature, CASE WHEN temperature > 30 THEN 'HIGH' WHEN temperature < 10 THEN 'LOW' ELSE 'NORMAL' END as temp_level FROM stream WHERE temperature > 20",
    "debug": true
  }
}
1
2
3
4
5
6
7
8
9

# String Processing

{
  "id": "s3",
  "type": "x/streamTransform",
  "name": "Device Information Formatting",
  "configuration": {
    "sql": "SELECT UPPER(deviceId) as device_id, CONCAT(location, '-', deviceType) as device_info, ROUND(temperature, 2) as temp FROM stream",
    "debug": false
  }
}
1
2
3
4
5
6
7
8
9

# Analytic Functions (Cross-Event State)

streamTransform processes each event synchronously, making it a natural fit for analytic functions that preserve state across events — each message is evaluated on arrival, and state is retained across messages for the node instance's lifetime.

-- CDC change detection: output when current crosses 300A from below
SELECT current, deviceId FROM stream
WHERE current > 300 AND lag(current) OVER (PARTITION BY deviceId) < 300

-- Output only when temperature changes
SELECT ts, temperature FROM stream WHERE had_changed(true, temperature) == true

-- Send only the changed fields, with a c_ prefix
SELECT changed_cols("c_", true, temperature, humidity) FROM stream

-- Lifetime accumulation (not reset by windows)
SELECT acc_sum(power) AS total, acc_count(*) AS cnt FROM stream
1
2
3
4
5
6
7
8
9
10
11
12

"No change" goes to the Filtered chain

When changed_cols/changed_col is the sole output and nothing changed this time, the node returns nil → routed to the Filtered chain (match=false). This is the intended event compression — wire the Filtered chain to drop/ignore; it never pollutes the Failure chain.

For full syntax and more usage (OVER/PARTITION BY/WHEN, conditional accumulation, etc.) see Analytic Functions.

# Stream-Table JOIN (Metadata Enrichment)

Configure tables to load metadata tables (device→location, product→category, etc.) and JOIN them in SQL to enrich stream rows. Tables can be loaded from file/http and refreshed periodically; the JOIN index key is auto-derived from the ON clause.

{
  "id": "enrich", "type": "x/streamTransform", "name": "Device Enrichment",
  "configuration": {
    "sql": "SELECT deviceId, m.location, m.type FROM stream s LEFT JOIN meta m ON s.deviceId = m.deviceId WHERE s.temp > 30",
    "tables": [
      {"name": "meta", "source": "file", "path": "/etc/rulego/device_meta.json", "format": "json", "refresh": "30s"}
    ]
  }
}
1
2
3
4
5
6
7
8
9

For the full tables fields and JOIN syntax see SQL Reference.

# Application Examples

# Example 1: IoT Data Preprocessing

Scenario: Clean and format raw data reported by IoT devices.

Rule Chain Configuration:

{
  "ruleChain": {
    "id": "iot_data_preprocessing",
    "name": "IoT Data Preprocessing",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamTransform",
        "name": "Data Cleaning",
        "configuration": {
          "sql": "SELECT deviceId, temperature, humidity, pressure, CASE WHEN temperature > 50 OR temperature < -20 THEN 'INVALID' ELSE 'VALID' END as data_quality FROM stream WHERE deviceId IS NOT NULL"
        }
      },
      {
        "id": "s2",
        "type": "jsFilter",
        "name": "Effective Data Filtering",
        "configuration": {
          "jsScript": "return msg.data_quality === 'VALID';"
        }
      },
      {
        "id": "s3",
        "type": "x/streamTransform",
        "name": "Unit Conversion",
        "configuration": {
          "sql": "SELECT deviceId, ROUND(temperature, 2) as temperature_c, ROUND(temperature * 1.8 + 32, 2) as temperature_f, ROUND(humidity, 1) as humidity_percent, pressure FROM stream"
        }
      },
      {
        "id": "s4",
        "type": "log",
        "name": "Processing Result",
        "configuration": {
          "jsScript": "return 'Processed: ' + JSON.stringify(msg);"
        }
      },
      {
        "id": "s5",
        "type": "log",
        "name": "Invalid Data",
        "configuration": {
          "jsScript": "return 'Invalid data: ' + JSON.stringify(msg);"
        }
      }
    ],
    "connections": [
      {"fromId": "s1", "toId": "s2", "type": "Success"},
      {"fromId": "s2", "toId": "s3", "type": "True"},
      {"fromId": "s2", "toId": "s5", "type": "False"},
      {"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

Input Data:

{"deviceId": "sensor001", "temperature": 25.678, "humidity": 65.432, "pressure": 1013.25}
1

Output Result:

{
  "deviceId": "sensor001",
  "temperature_c": 25.68,
  "temperature_f": 78.22,
  "humidity_percent": 65.4,
  "pressure": 1013.25
}
1
2
3
4
5
6
7

# Example 2: Data Standardization

Scenario: Standardize device data in different formats into a unified format.

Rule Chain Configuration:

{
  "ruleChain": {
    "id": "data_standardization",
    "name": "Data Standardization",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamTransform",
        "name": "Field Standardization",
        "configuration": {
          "sql": "SELECT UPPER(COALESCE(device_id, deviceId, id)) as device_id, COALESCE(temp, temperature, t) as temperature, COALESCE(hum, humidity, h) as humidity, CONCAT(COALESCE(location, 'unknown'), '-', COALESCE(building, 'default')) as location_info FROM stream"
        }
      },
      {
        "id": "s2",
        "type": "x/streamTransform",
        "name": "Data Classification",
        "configuration": {
          "sql": "SELECT *, CASE WHEN temperature > 25 AND humidity > 60 THEN 'HOT_HUMID' WHEN temperature > 25 THEN 'HOT_DRY' WHEN humidity > 60 THEN 'COOL_HUMID' ELSE 'COMFORTABLE' END as environment_type FROM stream"
        }
      },
      {
        "id": "s3",
        "type": "log",
        "name": "Standardization Result",
        "configuration": {
          "jsScript": "return 'Standardized: ' + JSON.stringify(msg);"
        }
      }
    ],
    "connections": [
      {"fromId": "s1", "toId": "s2", "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

# Example 3: Batch Processing of Array Data

Scenario: Process an array message containing multiple sensors' data for batch temperature conversion and filtering.

Input Data:

[
  {"sensorId": "s001", "value": 23.5, "unit": "C", "status": "active"},
  {"sensorId": "s002", "value": 45.2, "unit": "C", "status": "active"},
  {"sensorId": "s003", "value": 18.7, "unit": "C", "status": "inactive"},
  {"sensorId": "s004", "value": 35.8, "unit": "C", "status": "active"}
]
1
2
3
4
5
6

Rule Chain Configuration:

{
  "ruleChain": {
    "id": "batch_transform",
    "name": "Batch Data Transformation",
    "root": true
  },
  "metadata": {
    "nodes": [
      {
        "id": "s1",
        "type": "x/streamTransform",
        "name": "Batch Temperature Conversion",
        "configuration": {
          "sql": "SELECT sensorId, value as celsius, ROUND(value * 1.8 + 32, 1) as fahrenheit, CASE WHEN value > 30 THEN 'HIGH' ELSE 'NORMAL' END as temp_status FROM stream WHERE status = 'active'"
        }
      },
      {
        "id": "s2",
        "type": "log",
        "name": "Transform Success",
        "configuration": {
          "jsScript": "return 'Transformed ' + metadata.getValue('transformedCount') + ' out of ' + metadata.getValue('originalCount') + ' items: ' + JSON.stringify(msg);"
        }
      },
      {
        "id": "s3",
        "type": "log",
        "name": "Transform Failure",
        "configuration": {
          "jsScript": "return 'Failed to transform: ' + metadata.getValue('failedCount') + ' out of ' + metadata.getValue('originalCount') + ' items';"
        }
      }
    ],
    "connections": [
      {"fromId": "s1", "toId": "s2", "type": "Success"},
      {"fromId": "s1", "toId": "s3", "type": "Failure"}
    ]
  }
}
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

Success Chain Output:

[
  {"sensorId": "s001", "celsius": 23.5, "fahrenheit": 74.3, "temp_status": "NORMAL"},
  {"sensorId": "s002", "celsius": 45.2, "fahrenheit": 113.4, "temp_status": "HIGH"},
  {"sensorId": "s004", "celsius": 35.8, "fahrenheit": 96.4, "temp_status": "HIGH"}
]
1
2
3
4
5

Message Metadata:

{
  "match": "true",
  "originalCount": "4",
  "transformedCount": "3",
  "failedCount": "1"
}
1
2
3
4
5
6

Processing Notes

  • The original array contains 4 elements.
  • The WHERE condition filtered out sensor s003 with status 'inactive'.
  • The final output array contains 3 successfully transformed elements.
  • Metadata records detailed processing statistics.

# Example 4: CDC Change Detection (Current Crossing Threshold)

Scenario: Output when a device's current goes from low to high and crosses 300A (current > 300 and the previous value < 300).

Node Configuration:

{
  "id": "s4",
  "type": "x/streamTransform",
  "configuration": {
    "sql": "SELECT current, deviceId, ts FROM stream WHERE current > 300 AND lag(current) OVER (PARTITION BY deviceId) < 300"
  }
}
1
2
3
4
5
6
7

Input/Output:

{current:200, deviceId:1, ts:2}
{current:500, deviceId:1, ts:3}  → output (200→500 crosses threshold)
{current:600, deviceId:1, ts:4}  → not output (500→600 does not cross)
1
2
3

PARTITION BY deviceId gives each device its own "previous current"; lag is evaluated before WHERE so that WHERE can reference it.

# Example 5: Event Compression (Send Only Changed Fields)

Scenario: Upstream bandwidth is limited; each message carries only the fields that changed, with a c_ prefix.

Node Configuration:

{
  "id": "s5",
  "type": "x/streamTransform",
  "configuration": {
    "sql": "SELECT changed_cols(\"c_\", true, temperature, humidity) FROM stream"
  }
}
1
2
3
4
5
6
7

Input/Output:

{temperature:23, humidity:50} → {c_temperature:23, c_humidity:50} (first time, all change)
{temperature:23, humidity:55} → {c_humidity:55} (only humidity changes)
{temperature:23, humidity:55} → goes to the Filtered chain (nothing changed, row suppressed)
1
2
3

No change goes to the Filtered chain

When changed_cols is the sole output and nothing changed, it returns nil → routed to the Filtered chain (match=false). This is the intended event compression — wire the Filtered chain to drop/ignore.

# Example 6: Lifetime Accumulation (Statistics Since Boot)

Scenario: Track total energy, peak, and sample count since boot — not reset by windows, accumulated per event.

Node Configuration:

{
  "id": "s6",
  "type": "x/streamTransform",
  "configuration": {
    "sql": "SELECT acc_sum(power) AS total, acc_max(power) AS peak, acc_count(*) AS cnt, acc_avg(power) AS avg_power FROM stream"
  }
}
1
2
3
4
5
6
7

Each event accumulates on arrival and outputs the current accumulated value; state is retained for the node's lifetime.

# Notes

  1. SQL Syntax Restriction: Only non-aggregate queries are supported; cannot contain GROUP BY / aggregation functions (COUNT/SUM/AVG, etc.); analytic functions (lag/changed_col/acc_*, etc.) are not aggregations and are allowed.
  2. Data Type: Only JSON data type input is supported.
  3. Synchronous Processing: Transformation is synchronous and blocks the current message.
  4. Array Processing:
    • Each element in the array is processed by SQL transformation one by one.
    • Only elements that are successfully transformed and meet the WHERE condition are included in the output array.
    • Partial element failures do not affect the overall result, only the final array element count.
    • Metadata automatically adds processing statistics: originalCount, transformedCount, failedCount.
  5. WHERE Condition: Data that does not meet the WHERE condition is filtered out (routed to the Filtered chain) and excluded from the output.
  6. Performance: For large arrays, consider the impact of data volume on processing performance.
Edit this page on GitHub (opens new window)
Last Updated: 2026/07/11, 11:54:34
Stream Aggregator
File Component

← File Component→

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