技术分享 | EdgeX 规则引擎 eKuiper 实战

关于2022 EdgeX中国挑战赛
2022 EdgeX中国挑战赛暨中关村国际前沿科技创新大赛EdgeX专题赛正式拉开帷幕。大赛由北京市科委、中关村管委会指导，由Linux基金会主办，由阿里云、百度智能云、EMQ、GSMA 5G IN、英特尔、InnoSpace、中科创达、VMware、紫竹ET孵化器等联合承办单位共同支持，将在线上线下同步开赛。
本次大赛分设两大赛道：医疗、教育、消费行业赛道和能源、工业、供应链赛道。大赛致力于构建一个物联网及边缘计算的学习和分享平台，基于EdgeX Foundry，针对不同赛道的多个应用场景，以共享技术投资解决行业技术问题。除了40w+人民币总奖励，参赛者还可以获得技术、资本、市场等一系列参赛回报。

为帮助参赛选手更好地了解并运用相关技术，本次大赛将在7月至9月持续开展3轮技术培训，涵盖初、中、高不同层级，帮助开发者系统学习智能边缘系统知识。我们邀请到来自英特尔、VMware、阿里云等多家机构的技术专家进行分享。

中级培训第五场由EMQ边缘计算领域资深技术专家黄济泳为我们带来。本次培训的主题为：EdgeX规则引擎eKuiper实战。本篇推文是中级培训系列课程回顾的第四辑。（点此回顾：中级培训第三场-EdgeX应用落地-零售、能源行业）

1. eKuiper简介与快速启动

1.1 eKuiper简介

图注：eKuiper引擎技术架构

eKuiper是Linux基金会旗下LF Edge的Stage1项目，设计目的为轻量物联网边缘数据流处理引擎。

eKuiper具有以下一系列优势与特点：
超轻量：10MB级别，足够小，可以部署在边缘网关甚至端设备上
快速：就近处理，超低延时通用计算，大吞吐量
整合：承上启下，连接各种协议的南桥北桥
进化：高可扩展性，轻松适配外部生态和算法

图注：eKuiper引擎功能结构

eKuiper根据其功能架构，具备以下四点主要特性。
基于SQL/Flow编辑器的业务规划
丰富的源和目标支持（内置标准MQTT消息服务器，EdgeX消息总线，定时拉取HTTP服务）
易扩展（支持插件机制，支持流定义函数（UDF）等等）
敏捷运维（支持各种平台二进制，规则热创建，热更新）

eKuiper是EdgeX Foundry的默认规则引擎。根据EdgeX的主要应用场景（通用工业IOT边缘物联网设备），eKuiper的主要业务即表现为：

接入EdgeX Foundry消息总线，进行数据分析和导出
作为EdgeX Foundry规则引擎，支持场景联动，按规则控制设备
写入EdgeX Foundry，起到类似设备的作用

1.2 eKuiper快速启动

图注：eKuiper快速启动代码

官方docker compose
- docker compose up -d
配置
- 文件etc目录下
- 环境变量
Image
- alpine
- 需要插件用slim
- slim-python

在此基础上，黄济泳专家为我们带来了两个eKuiper应用示例Demo。
图注：Demo 0：变更启动镜像
使用slim版本（可使用1.6.1版本之前预编译插件）
添加eKuiper manager（更新更快的图形界面）

图注：Demo 1：创建第一个规则
使用EdgeX UI
使用eKuiper manager
（http://edgex-kuiper:59720）

2. eKuiper处理EdgeX消息实战

2.1 eKuiper与EdgeX的集成

eKuiper通过三种方式与EdgeX集成通信：

接入EdgeX消息总线（提供EdgeX Source，自动解析数据）
发送EdgeX消息总线（提供EdgeX Sink，传输结果到总线）
规则触发EdgeX服务，如触发command服务控制设备
图注：EdgeX到eKuiper的数据流动

EdgeX到eKuiper数据流通：

默认通过app-service-rules
可配置app service的变换
可配置多stream

详细教程：
https://ekuiper.org/docs/en/latest/edgex/edgex_source_tutorial.html

图注：利用app service按照设备过滤数据

Source配置流程如下

eKuiper中edgex.yaml配置
- 可通过docker compose配置
创建处理Random-Integer-Device设备的流

CREATE STREAM edgeInt() with (FORMAT = "JSON", TYPE = "edgex", CONF_KEY = "device_conf")

图注：source配置示例，默认选择edgex总线

图注：eKuiper到EdgeX使用场景示例

eKuiper到EdgeX使用场景：

扩展EdgeX数据来源，连接异构系统
- MQTT→eKuiper→EdgeX
- HTTP→eKuiper→EdgeX
- Private protocol→eKuiper→EdgeX
消息处理完后，Republish
配置元数据：profile，device etc.

图注：Demo：MQTT消息导入EdgeX
流：MQTT topic demo的数据，模拟外部数据
规则逻辑：采集输入中的温湿度数据
输出：类似device数据发送至EdgeX中

eKuiper规则输出
可同时输出到多个目的，例如：MQTT、REST
格式化利器：DataTemplate

2.2 eKuiper数据分析能力

时间窗口

Tumbling / Hopping / Sliding / Session / Count
多流Join

丰富内置函数

数据去重、数字、字符、转换和hash等函数
JSON Path

常见SQL语句支持

WHERE
CASE WHEN
GROUP BY
ORDER BY
HAVING

2.3 eKuiper扩展能力

图注：eKuiper外部扩展能力

支持的扩展类型

Go原生
外部服务定义
自研多语言插件框架

Go原生插件痛点

编译和运行诸多限制
无法热插拔

Portable插件框架

支持Source/Sink/Function
支持Go/Python
可扩展到更多语言
基于NNG协议，IPC通信

————————————————

最后，让我们一起来看看今天的练习题吧~ （答案于文章最后公布）

01（多选）下列关于eKuiper的说法正确的是？
A、EdgeX Foundry的第三方微服务
B、由EdgeX Foundry团队开发
C、开源软件
D、闭源商业软件
E、与EdgeX Foundry同样属于LF Edge基金会
F、提供免费UI

02（多选）eKuiper如何与EdgeX整合？
A、可读取EdgeX消息总线
B、可读取EdgeX数据库
C、可读取 EdgeX app service 的输出
D、可写入 EdgeX 消息总线
E、可调用 command service 控制设备
F、可作为device service

03（多选）eKuiper能做什么？
A、数据清洗
B、实时分析
C、过滤和告警
D、持续存储
E、图像处理
F、AI推理

————————————————

2022 EdgeX中国挑战赛已于8月3日盛大开幕，EdgeX中文社区将在接下来的日子里为大家带来更多边缘计算讲座分享和赛事进度更新。在这个盛夏，让我们一起展现风采，从自身所学的领域出发，为物联网、边缘计算市场提供更多更好的解决方案，展望一个更好的科技未来。

第三届EdgeX中国挑战赛，邀您来赛

多场景落地前景可期
全链条助阵生态共建
阶梯式培训同期推进
大奖与荣誉虚位以待

2022EdgeX中国挑战赛正在火热报名中，报名时间：2022年8月3日00:00 - 9月5日24:00。参与大赛，共同推动万物互联，请即刻联系我们。

大赛官网：https://www.edgex-challenge.com/

✦
————————————————
答案： 1. ACEF 2.ACDE 3. ABCEF