使用 LF Edge eKuiper 将物联网流处理数据写入 Databend

2023-10-19
福建
本文字数：2192 字
阅读完需：约 7 分钟

作者：韩山杰
Databend Cloud 研发工程师
https://github.com/hantmac

LF Edge eKuiper

LF Edge eKuiper 是 Golang 实现的轻量级物联网边缘分析、流式处理开源软件，可以运行在各类资源受限的边缘设备上。eKuiper 的主要目标是在边缘端提供一个流媒体软件框架（类似于 Apache Flink (opens new window)）。eKuiper 的规则引擎允许用户提供基于 SQL 或基于图形（类似于 Node-RED）的规则，在几分钟内创建物联网边缘分析应用。具体介绍可以参考 [LF Edge eKuiper - 超轻量物联网边缘流处理软件(https://ekuiper.org/docs/zh/latest/)。

Databend Sql Sink

eKuiper 支持通过 Golang 或者 Python 在源 (Source)，SQL 函数, 目标 (Sink) 三个方面的扩展，通过支持不同的 Sink，允许用户将分析结果发送到不同的扩展系统中。Databend 作为 Sink 也被集成到了 eKuiper plugin 当中，下面通过一个案例来展示如何使用 eKuiper 将物联网流处理数据写入 Databend。

编译 eKuiper 和 Databend Sql Plugin

eKuiper

git clone https://github.com/lf-edge/ekuiper & cd ekuipermake

复制代码

Databend Sql Plugin

go build -trimpath --buildmode=plugin -tags databend -o plugins/sinks/Sql.so extensions/sinks/sql/sql.go

复制代码

编译后的 sink plugin 拷贝到 build 目录：

cp plugins/sinks/Sql.so _build/kuiper-1.11.1-18-g42d9147f-darwin-arm64/plugins/sinks

复制代码

Databend 建表

在 Databend 中先创建目标表 ekuiper_test：

create table ekuiper_test (name string,size bigint,id bigint);

复制代码

启动 eKuiperd

cd _build/kuiper-1.11.1-18-g42d9147f-darwin-arm64 ./bin/kuiperd

复制代码

服务正常启动：

创建流（stream）和规则 (rule)

eKuiper 提供了两种管理各种流、规则，目标端的方式，一种是通过 ekuiper-manager 的 [docker image]（https://hub.docker.com/r/lfedge/ekuiper）启动可视化管理界面，一种是通过 CLI 工具来管理。这里我们使用 CLI。

创建 stream

流是 eKuiper 中数据源连接器的运行形式。它必须指定一个源类型来定义如何连接到外部资源。这里我们创建一个流，从 json 文件数据源中获取数据，并发送到 eKuiper 中。

首先配置文件数据源，连接器的配置文件位于 /etc/sources/file.yaml。

default:  # 文件的类型，支持 json， csv 和 lines  fileType: json  # 文件以 eKuiper 为根目录的目录或文件的绝对路径。  # 请勿在此处包含文件名。文件名应在流数据源中定义  path: data  # 读取文件的时间间隔，单位为ms。如果只读取一次，则将其设置为 0  interval: 0  # 读取后，两条数据发送的间隔时间  sendInterval: 0  # 是否并行读取目录中的文件  parallel: false  # 文件读取后的操作  # 0: 文件保持不变  # 1: 删除文件  # 2: 移动文件到 moveTo 定义的位置  actionAfterRead: 0  # 移动文件的位置, 仅用于 actionAfterRead 为 2 的情况  moveTo: /tmp/kuiper/moved  # 是否包含文件头，多用于 csv。若为 true，则第一行解析为文件头。  hasHeader: false  # 定义文件的列。如果定义了文件头，该选项将被覆盖。  # columns: [id, name]  # 忽略开头多少行的内容。  ignoreStartLines: 0  # 忽略结尾多少行的内容。最后的空行不计算在内。  ignoreEndLines: 0  # 使用指定的压缩方法解压缩文件。现在支持`gzip`、`zstd` 方法。  decompression: ""

复制代码

使用 CLI 创建 steam 名为 stream1:

./bin/kuiper create stream stream1 '(id BIGINT, name STRING,size BIGINT) WITH (DATASOURCE="test.json", FORMAT="json", TYPE="file");'

复制代码

Json 文件的内容为：

[  {"id": 1,"size":100, "name": "John Doe"},  {"id": 2,"size":200, "name": "Jane Smith"},  {"id": 3,"size":300, "name": "Kobe Brant"},  {"id": 4,"size":400, "name": "Alen Iverson"}]

复制代码

创建 Databend Sink Rule

一个规则代表了一个流处理流程，定义了从将数据输入流的数据源到各种处理逻辑，再到将数据输入到外部系统的动作。eKuiper 有两种方法来定义规则的业务逻辑。要么使用 SQL / 动作组合，要么使用新增加的图 API。

这里我们通过指定 sql 和 actions 属性，以声明的方式定义规则的业务逻辑。其中，sql 定义了针对预定义流运行的 SQL 查询，这将转换数据。然后，输出的数据可以通过 action 路由到多个位置。

规则由 JSON 定义，下面是准备创建的规则 myRule.json：

{  "id": "myRule",  "sql": "SELECT id, name from stream1",  "actions": [    {      "log": {      },      "sql": {        "url": "databend://databend:databend@localhost:8000/default?sslmode=disable",        "table": "ekuiper_test",        "fields": ["id","name"]      }    }  ]}

复制代码

执行 CLI 创建规则：

./bin/kuiper create rule myRule -f myRule.json

复制代码

可以查看所创建规则的运行状态：

./bin/kuiper getstatus rule myRule

复制代码

规则创建后，会立即将符合规则条件的数据发送到目标端，此时我们查看 Databend 的 ekuiper_test 表，可以看到文件数据源中的数据已经被写入到 Databend：

可以看到由于我们的规则 SQL 中只指定了 id, name 字段，所以这里只有这两个字段被写入。

结论

eKuiper 是 EMQ 旗下的一款流处理软件，其体积小、功能强大，在工业物联网、车辆网、公共数据分析等很多场景中得到广泛使用。本文介绍如何使用 eKuiper 将物联网流处理数据写入 Databend。

发布于: 刚刚阅读数: 4

原文链接:【http://xie.infoq.cn/article/f4c8194fe8a89e4f65ac99d2e】。文章转载请联系作者。

Databend

关注

还未添加个人签名 2022-08-25 加入

还未添加个人简介

发布

暂无评论

创作场景

使用 LF Edge eKuiper 将物联网流处理数据写入 Databend

LF Edge eKuiper

Databend Sql Sink

编译 eKuiper 和 Databend Sql Plugin

eKuiper

Databend Sql Plugin

Databend 建表

启动 eKuiperd

创建流（stream） 和 规则 (rule)

创建 stream

创建 Databend Sink Rule

结论

Databend

评论

创建流（stream）和规则 (rule)