运行环境 centos7
官网下载rpm包 并安装
https://ekuiper.org/zh/downloads-and-install?version=1.13.6&os=Linux&oslable=Linux
ekuiper流计算的方式类似flink table api那一套 熟悉的比较快能上手
首先配置mqtt source源
通过内置的yaml配置
vim /etc/kuiper/mqtt_source.yaml
#Global MQTT configurations
default:
qos: 0
server: "tcp://127.0.0.1:1883"
protocolVersion: "3.1.1"
insecureSkipVerify: false
#decompression: zlib
username: admin
password: public
#certificationPath: /var/kuiper/xyz-certificate.pem
#privateKeyPath: /var/kuiper/xyz-private.pem.key
#rootCaPath: /var/kuiper/xyz-rootca.pem
#insecureSkipVerify: false
#connectionSelector: mqtt.mqtt_conf1
#kubeedgeVersion:
#kubeedgeModelFile: ""
# demo_conf: #Conf_key
# qos: 0
# server: "tcp://10.211.55.6:1883"
创建一个流（类似flink创建一个table）
/usr/bin/kuiper create stream demo '(temperature float, humidity bigint) WITH (FORMAT="JSON", DATASOURCE="demo")'
DATASOURCE配置的是要读取的topic 其实还有个type属性 用来指定源类型 不配置默认是mqtt 所以此处source为mqtt
流创建以后 是惰性的 在没有执行sql查询 或者启动一个规则之前 流是不会运作的 此时demo流还没有开始读取demo topic
执行一个sql 看看效果
/usr/bin/kuiper query
kuiper > select count(*), avg(humidity) as avg_hum, max(humidity) as max_hum from demo where temperature > 30 group by TUMBLINGWINDOW(ss, 30);
Query was submit successfully.
这个sql的意思是 通过每30s的滚动窗口 统计30度以上的平均湿度和最高湿度
往mqtt demo topic推送消息
{ "temperature": 35,"humidity":60 }
{ "temperature": 35,"humidity":70 }
可以看到控制台打印出流计算的结果
kuiper > [{}]
[{"avg_hum":65,"count":2,"max_hum":70}]
此时已经完成了一个流计算 ctrl+c退出query
配置一个规则 把流计算结果输出到sink
首先配置一个规则文件 rule1.txt
vim /usr/ekuiper/rule1.txt
{
"sql": "SELECT avg(humidity) as avg_hum,format_time(tstamp(),'YYYY-MM-dd HH:mm:ss') as current_time from demo where temperature > 30 group by TUMBLINGWINDOW(ss, 30)",
"actions": [
{
"log": {}
},
{
"mqtt": {
"server": "tcp://127.0.0.1:1883",
"qos": 0,
"username": "admin",
"password": "public",
"topic": "demoSink"
}
}
]
}
action里面配置的就是sink 该规则就是把流计算结果发送到demoSink topic
创建规则
/usr/bin/kuiper create rule rule1 -f /usr/ekuiper/rule1.txt
规则创建出来以后 默认就是运行状态
查看规则的状态
/usr/bin/kuiper getstatus rule rule1
{
"status": "running",
"lastStartTimestamp": "1720686733949",
"lastStopTimestamp": "0",
"nextStopTimestamp": "0",
"source_demo_0_records_in_total": 5,
"source_demo_0_records_out_total": 5,
"source_demo_0_messages_processed_total": 5,
"source_demo_0_process_latency_us": 9,
"source_demo_0_buffer_length": 0,
"source_demo_0_last_invocation": "2024-07-11T17:49:41.150715",
"source_demo_0_exceptions_total": 0,
"source_demo_0_last_exception": "",
"source_demo_0_last_exception_time": 0,
...
往demo topic里推送数据 再消费demoSink topic
可以看到demoSink里面已经有数据了
数据推送
{ "temperature": 35,"humidity":50 }
{ "temperature": 35,"humidity":60 }
{ "temperature": 35,"humidity":70 }
消费demoSink
[{"avg_hum":60,"current_time":"2024-07-11 18:11:30"}]
至此实现了一个完整的流计算
eKuiper是物联网领域的流计算框架
文档地址 https://ekuiper.org/docs/zh/latest/
