Quick Start

# Quick Start

This guide will take you through StreamSQL's basic features in 5 minutes, from installation to running your first stream processing program.

# Environment Requirements

Go 1.18 or higher
Basic Go language development experience
Understanding of basic SQL syntax (optional, but helpful for understanding)

# Installation

# 1. Create New Project

mkdir my-streamsql-app
cd my-streamsql-app
go mod init my-streamsql-app

1
2
3

# 2. Add Dependencies

go get github.com/rulego/streamsql

# 3. Verify Installation

Create a simple test file to verify installation:

package main

import (
    "fmt"
    "github.com/rulego/streamsql"
)

func main() {
    ssql := streamsql.New()
    fmt.Println("StreamSQL installed successfully!")
    ssql.Stop()
}

1
2
3
4
5
6
7
8
9
10
11
12

# Core Concepts Overview

Before starting to write code, understand a few core concepts:

Stream: Continuous data sequence, similar to a table in a database
Window: Mechanism to divide unbounded streams into bounded datasets
Aggregation: Statistical calculations on data within windows
Sink: Callback function to process query results

# First StreamSQL Program

# 1. Basic Example - Simple Data Filtering

This example shows how to filter real-time data streams:

Create main.go file:

package main

import (
    "fmt"
    "time"
    "github.com/rulego/streamsql"
)

func main() {
    // 1. Create StreamSQL instance
    ssql := streamsql.New()
    defer ssql.Stop()
    
    // 2. Define SQL query - filter data with temperature greater than 25 degrees
    sql := "SELECT deviceId, temperature FROM stream WHERE temperature > 25"
    
    // 3. Execute SQL query
    err := ssql.Execute(sql)
    if err != nil {
        panic(err)
    }
    
    // 4. Add result processing function
    ssql.AddSink(func(results []map[string]interface{}) {
        fmt.Printf("High temperature alert: %v\n", results)
    })
    
    // 5. Send test data
    testData := []map[string]interface{}{
        {"deviceId": "sensor001", "temperature": 23.5}, // Won't trigger alert
        {"deviceId": "sensor002", "temperature": 28.3}, // Will trigger alert
        {"deviceId": "sensor003", "temperature": 31.2}, // Will trigger alert
    }
    
    for _, data := range testData {
        ssql.Emit(data)
        time.Sleep(100 * time.Millisecond)
    }
    
    // Wait for processing to complete
    time.Sleep(1 * time.Second)
}

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

Run the program:

go run main.go

Expected output:

High temperature alert: [map[deviceId:sensor002 temperature:28.3]]
High temperature alert: [map[deviceId:sensor003 temperature:31.2]]

1
2

Code Analysis:

streamsql.New() - Create StreamSQL instance
Execute(sql) - Parse and execute SQL query
AddSink() - Register result processing function
Emit() - Add data to the stream
WHERE temperature > 25 - Filter condition, only process data with temperature greater than 25 degrees

Important Note: For aggregate queries (using window functions), you need to wait for the window time to arrive or manually call ssql.TriggerWindow() to trigger window calculations.

EmitSync for Non-Aggregation Queries

The filter above is a non-aggregation query. Besides Emit + AddSink, you can use EmitSync to get a single result synchronously, skipping the callback and time.Sleep:

result, err := ssql.EmitSync(map[string]any{"deviceId": "sensor002", "temperature": 28.3})
// result == {deviceId:sensor002 temperature:28.3}; when WHERE is not matched, result == nil (not an error)

1
2

Non-aggregation (filtering / transformation / analytic functions) → EmitSync; aggregation / windows → Emit + AddSink. See Core Concepts · Two Execution Paths for both APIs.
For change detection (CDC), lag, cumulative statistics, etc., see Analytic Functions.

# 2. Aggregation Analysis Example - Calculate Average Temperature

package main

import (
    "fmt"
    "math/rand"
    "time"
    "github.com/rulego/streamsql"
)

func main() {
    ssql := streamsql.New()
    defer ssql.Stop()
    
    // Calculate average temperature for each device every 5 seconds
    sql := `SELECT deviceId, 
                   AVG(temperature) as avg_temp,
                   COUNT(*) as sample_count,
                   window_start() as window_start,
                   window_end() as window_end
            FROM stream 
            GROUP BY deviceId, TumblingWindow('5s')`
    
    err := ssql.Execute(sql)
    if err != nil {
        panic(err)
    }
    
    // Process aggregation results
    ssql.AddSink(func(results []map[string]interface{}) {
        fmt.Printf("Aggregation results: %v\n", results)
    })
    
    // Simulate sensor data stream
    devices := []string{"sensor001", "sensor002", "sensor003"}
    for i := 0; i < 8; i++ {
        for _, device := range devices {
            data := map[string]interface{}{
                "deviceId":    device,
                "temperature": 20.0 + rand.Float64()*15, // Random temperature 20-35 degrees
                "timestamp":   time.Now(),
            }
            ssql.Emit(data)
        }
        time.Sleep(300 * time.Millisecond)
    }
    
    // Wait for window to trigger
    time.Sleep(5 * time.Second)
    time.Sleep(500 * time.Millisecond)
}

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

# Advanced Examples

# 3. Sliding Window Analysis

package main

import (
    "fmt"
    "math/rand"
    "time"
    "github.com/rulego/streamsql"
)

func main() {
    ssql := streamsql.New()
    defer ssql.Stop()
    
    // 30-second sliding window, sliding every 10 seconds
    sql := `SELECT deviceId,
                   AVG(temperature) as avg_temp,
                   MAX(temperature) as max_temp,
                   MIN(temperature) as min_temp
            FROM stream 
            WHERE temperature > 0
            GROUP BY deviceId, SlidingWindow('30s', '10s')`
    
    err := ssql.Execute(sql)
    if err != nil {
        panic(err)
    }
    
    ssql.AddSink(func(results []map[string]interface{}) {
        fmt.Printf("Sliding window analysis: %v\n", results)
    })
    
    // Continuously send data
    for i := 0; i < 10; i++ {
        data := map[string]interface{}{
            "deviceId":    "sensor001",
            "temperature": 20.0 + rand.Float64()*10,
            "timestamp":   time.Now(),
        }
        ssql.Emit(data)
        time.Sleep(800 * time.Millisecond)
    }
    
    time.Sleep(1 * time.Second)
}

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

# 4. Nested Field Access Example

package main

import (
    "fmt"
    "time"
    "github.com/rulego/streamsql"
)

func main() {
    ssql := streamsql.New()
    defer ssql.Stop()
    
    // SQL query to access nested fields
    sql := `SELECT device.info.name as device_name,
                   device.location.building as building,
                   sensor.temperature as temp,
                   UPPER(device.info.type) as device_type
            FROM stream 
            WHERE sensor.temperature > 25 AND device.info.status = 'active'`
    
    err := ssql.Execute(sql)
    if err != nil {
        panic(err)
    }
    
    ssql.AddSink(func(results []map[string]interface{}) {
        fmt.Printf("Nested field results: %v\n", results)
    })
    
    // Send nested structure data
    complexData := map[string]interface{}{
        "device": map[string]interface{}{
            "info": map[string]interface{}{
                "name":   "Temperature Sensor 001",
                "type":   "temperature",
                "status": "active",
            },
            "location": map[string]interface{}{
                "building": "Building A",
                "floor":    "3F",
            },
        },
        "sensor": map[string]interface{}{
            "temperature": 28.5,
            "humidity":    65.0,
        },
    }
    
    ssql.Emit(complexData)
    time.Sleep(500 * time.Millisecond)
}

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

Advanced Usage

Custom functions (UDF), performance modes (high throughput / low latency), and other advanced usage are covered in Advanced Examples.

# Performance Tips

Choose appropriate window size: Too small windows increase computational overhead, too large windows increase memory usage
Use filtering conditions reasonably: Filter data early in WHERE clauses to improve performance
Avoid complex nested queries: StreamSQL is optimized for simple and efficient queries
Monitor memory usage: Monitor memory usage in high-frequency data scenarios

# Common Questions

# Q: No output results for data?

A: Check the following:

Ensure AddSink() is called to add result processing function
If using window functions, ensure window is triggered (time elapsed or manually triggered)
Check if WHERE conditions filter out all data

# Q: When do window functions trigger?

Tumbling window: Automatically triggers when window end time is reached
Sliding window: Triggers once per sliding interval
Counting window: Triggers when the count reaches the specified number
Session window: Triggers after the session timeout

# Q: How to handle abnormal data?

A: Use the WHERE clause to filter abnormal data:

SELECT * FROM stream
WHERE temperature IS NOT NULL
  AND temperature BETWEEN -50 AND 100

1
2
3

# RuleGo Integration Example

# Complete data processing pipeline

// Receive MQTT data via RuleGo, integrate StreamSQL in a rule chain, then output
ruleChain := `{"ruleChain":{"nodes":[{"id":"s1","type":"streamSqlNode","configuration":{"sql":"SELECT device_id, AVG(temperature) as avg_temp FROM stream GROUP BY device_id, TumblingWindow('5m')"}},{"id":"s2","type":"restApiCallNode","configuration":{"restEndpointUrlPattern":"http://alert-service/api/alerts","requestMethod":"POST"}}],"connections":[{"fromId":"s1","toId":"s2","type":"Success"}]}}`
engine := rulego.New([]byte(ruleChain))
msg := rulego.NewMsg(0, "deviceData", types.JSON, nil, `{"device_id": "dev1", "temperature": 25.5}`)
engine.OnMsg(msg)

1
2
3
4
5

# Advantages

Data ingestion: RuleGo MQTT/HTTP/WebSocket components
Stream processing: StreamSQL for real-time analytics
Output: databases, APIs, message queues via RuleGo output components
Rule management: dynamic config and hot-update
Monitoring: RuleGo monitoring and alerting

# Complete Example Code

All example code is in the project examples/ directory:

examples/simple-custom-functions/ (opens new window) - basic usage
examples/custom-functions-demo/ (opens new window) - full feature demo
examples/function-integration-demo/ (opens new window) - integration use case
examples/advanced-functions/ (opens new window) - advanced features
examples/comprehensive-test/ (opens new window) - comprehensive test demo for this tutorial

# Quick installation verification

cd examples/comprehensive-test
go run main.go

1
2

It covers all features in this doc and is the best starting point for verifying your installation.

Edit this page on GitHub

Last Updated: 2026/07/11, 04:40:00

← Overview Core Concepts→