EMQ 物聯網邊緣流式數據處理軟件 Kuiper 正式開源

時間 2020-01-12

原文原文鏈接

全球領先的開源物聯網軟件提供商 EMQ (Github 項目) 於2019年10月23日正式開源了一個邊緣端的超輕量級流式數據處理項目 EMQ X Kuiper（如下簡稱 Kuiper）。隨着 Kuiper 的發佈，EMQ 將加快在物聯網中間件領域的佈局，從而造成從消息接入到數據處理，從邊緣到雲端的物聯網中間件軟件全領域覆蓋。git

物聯網邊緣計算新需求

2019 年做爲 5G 商用元年，隨着 5G 部署的深刻，邊緣計算也已經日漸流行。邊緣計算經過在靠近接入網的機房增長計算能力，能夠，github

大幅下降業務時延
減小對網絡的帶寬壓力，下降傳輸成本
減小在雲端的數據存儲成本
提升內容分發效率提高用戶體驗

物聯網邊緣計算很大一部分指的是流數據的處理，所謂流數據是一組順序、大量、快速、連續到達的數據序列，通常狀況下，流數據可被視爲一個隨時間延續而無限增加的動態數據集合。流式數據處理能夠幫助用戶實時瞭解系統設備的狀態，並對異常狀況做出快速響應。在邊緣端的計算資源（CPU，內存等）不像在雲端豐富，所以傳統的流式數據處理框架相似於 Apache Storm 或者 Apache Flink 等，因爲其安裝包過大，或者部署結構與過程過於複雜，運行時的高消耗等緣由，並不適合於在這些資源受限的邊緣設備（工控機、網關，或者配置不高的 X86 服務器等設備）上運行。而 EMQ 開源的 Kuiper 就是爲了解決在物聯網邊緣設備上的這些問題。sql

Kuiper 特色

輕量級：安裝包約 7MB左右；不依賴於第三方的庫與運行時，下載解壓後便可運行，安裝和使用很是方便
跨操做系統支持：目前能夠運行在 Linux 和 Mac 上，後續版本已規劃會支持 Windows 操做系統
使用簡單，快速應對敏捷業務變化：用戶能夠經過寫 SQL 來實現業務處理；若是業務有變化，只需更新相關的 SQL 語句便可，而無需進行復雜的應用開發和部署
內置支持 MQTT 消息的處理：做爲物聯網領域的事實協議標準，Kuiper 內置了對接 EMQ X 等 MQTT 消息服務器，實現對來自於物聯網消息處理的無縫對接
可擴展性：Kuiper 定義了擴展接口，後續版本已經規劃擴展實現不一樣消息源處理和處理後數據的沉澱（sink）
可管理性：Kuiper 定義了管理接口，後續版本已經規劃能夠經過 REST API 來實現這些業務規則的定義和管理等操做

Kuiper 應用場景

Kuiper 能夠應用在各種邊緣計算的場景中，典型應用場景包括，shell

車間的工業網關：實時分析來自於生產線的數據，並將分析結果推送至雲端。車間本地可視化系統，或者遠程監控系統能夠查看生產線的實時狀態
車聯網車機：實時分析汽車總線的數據，通過分析將有價值的數據推送至雲端或者本地存儲，車載系統或者用戶的手機應用能夠實時查看汽車的狀態
智能家居網關：經過實時分析家居採集的各種數據，將重要的結果經過本地顯示設備，或者經過雲端發送給用戶的手機應用，實現對家庭設備的即時狀態管理與控制

以上各種應用場景很好體現了邊緣計算的價值：實時本地數據處理，提高用戶體驗；在本地數據處理，還提升了數據安全性，保護我的隱私；節省網絡帶寬和雲端存儲成本。Kuiper 能夠運行在這些計算資源不足的邊緣設備上，低成本地實現邊緣端的流式數據處理。數據庫

Kuiper 組件概述

以下圖所示，Kuiper Server 包括了三部分，分別爲安全

Engines：這部分爲核心的處理引擎，主要功能包括了 SQL 解析，SQL 運行時和流式數據處理框架等
RPC 接口：RPC 接口主要用於跟客戶端命令行工具的交互。包括流定義與管理，規則定義與管理，查詢工具等
Rest API 接口：Rest API 主要用於未來的管理接口，以及在計劃中的管理界面的交互。包括流定義與管理，規則定義與管理，查詢工具等

Kuiper 使用步驟概述

Kuiper 的使用過程與傳統的流式工具相似，主要包括三步：流定義，規則定義，提交規則執行服務器

流定義：這個步驟主要用於定義數據源流出的數據格式，相似於傳統關係型數據庫中定義表結構的過程。以下所示定義了一個名爲 demo 的數據流網絡
```
# bin/cli create stream demo '(temperature float, humidity bigint) WITH (FORMAT="JSON", DATASOURCE="demo")'
```
規則定義：該步驟主要用於定義如下三部份內容框架
- 數據源（Source）：這部分指定被處理的數據從何而來，在 SQL 的 FROM 語句中指定，就是在上一步中定義好的流的名稱
- 數據業務處理（Process）：這部分制定瞭如何處理數據，主要經過 SQL 語句的 SELECT、JOIN、GROUP、WINDOW（時間窗口），以及各類函數來實現
- 數據沉澱（Sink）：這部分指定在數據處理完成後，將處理後的數據沉澱至相應的目標。目前支持將數據沉澱到 log 和 MQTT 服務器
```
{
    "sql": "SELECT * from demo where temperature > 30",
    "actions": [{
        "log":  {}
    }]
}
```

如上所示，定義了一個規則，從名稱爲 demo 的流中，取出溫度大於30的數據，並將知足條件的數據輸出到日誌文件中。規則能夠保存在一個文本文件中，好比 myRule函數

提交規則執行

# bin/cli create rule ruleDemo -f myRule

完成規則後，經過建立規則命令就能夠將其提交。知足條件的數據將被輸出至日誌文件。

...
[{"humidity":17,"temperature":60}]
[{"humidity":59,"temperature":68}]
[{"humidity":67,"temperature":31}]
[{"humidity":5,"temperature":45}]
[{"humidity":34,"temperature":97}]
[{"humidity":90,"temperature":73}]
[{"humidity":72,"temperature":59}]
[{"humidity":39,"temperature":72}]
[{"humidity":76,"temperature":92}]
[{"humidity":23,"temperature":40}]
[{"humidity":74,"temperature":91}]
[{"humidity":50,"temperature":76}]
[{"humidity":77,"temperature":44}]
...