基于 Kuiper 中的 Golang 模板定制分析结果

2023-12-06 21:38:28

释放物联网数据分析潜能：掌握 Kuiper 的 Golang 模版

物联网 (IoT) 数据分析的复杂性正在日益增加，迫切需要一种简化数据分析和输出到不同系统的解决方案。Kuiper 流式数据分析引擎通过其创新的 Golang 模版功能，为这一挑战提供了优雅的解答。

模版：简化数据格式化

模版是 Kuiper 中的特殊函数，可以将分析结果按照指定的格式输出。这使得用户能够轻松地将数据发送到各种接收器，而无需为每个接收器编写单独的规则。模版函数的语法如下：

{{ template "template_name" . }}

其中，template_name 是模版的名称，. 代表当前规则的上下文，包含所有输入值。

内置模版

Kuiper 内置了几个常用的模版，包括：

json：将输入值转换为 JSON 格式
csv：将输入值转换为 CSV 格式
rest：将输入值发送到云端 REST 服务
mqtt：将输入值发送到本地 MQTT 服务器

自定义模版

用户还可以创建自己的自定义模版。只需定义一个 Golang 函数，并将其名称指定为模版名称即可。例如：

// my_template is a custom template function.
func myTemplate(ctx resource.RuleContext) (string, error) {
    // Get the input value.
    value := ctx.Streams[0].Rows[0].Values[0].Value

    // Convert the input value to a string.
    strValue := fmt.Sprintf("%v", value)

    // Return the formatted string.
    return strValue, nil
}

然后，可以在规则中使用此模版：

{{ template "my_template" .temperature }}

示例代码

以下代码示例演示了如何使用 Kuiper 模版将数据发送到 MQTT 服务器：

package main

import (
	"context"
	"fmt"
	"github.com/apache/kurier/client/pkg/resource"
)

func main() {
	// Create a new Kuiper client.
	client, err := resource.NewKuiperClient("127.0.0.1:8100")
	if err != nil {
		fmt.Println(err)
		return
	}

	// Create a new rule.
	rule := resource.Rule{
		SQL: `
SELECT
    temperature
FROM
    temperature_stream;
`,
		Sink: []resource.RuleSink{
			{
				Type: "mqtt",
				Config: map[string]interface{}{
					"address": "localhost:1883",
					"topic":   "temperature_data",
					// Use the "json" template to format the output.
					"template": `{{ template "json" . }}`,
				},
			},
		},
	}

	// Submit the rule to Kuiper.
	if _, err := client.Rules.Create(context.Background(), &rule); err != nil {
		fmt.Println(err)
		return
	}

	fmt.Println("Rule created successfully!")
}