一小時搭建 EMQ X + InfluxDB + Grafana 物聯網數據可視化方案

背景

本文以常見物聯網使用場景爲例，介紹瞭如何利用 EMQ X 消息中間件與開源數據可視化方案 InfluxDB + Grafana ，將物聯網設備大量基於時序的數據便捷地展現出來。

在物聯網項目中接入平臺的設備數據和存儲方案如下特色：

• 數據採集的維度，頻率，以及設備數量都比較多，採集的數據量比較大，對消息服務器接入點吞吐量、後端數據庫數據存儲空間消耗有很大壓力；
• 數據按照採集週期進行上報、傳輸、存儲是通常都按照時間序列；

所以在物聯網項目中使用時序數據庫是比較好的選擇。時序數據庫能夠帶來顯著的性能的提高，包括更高的容納率、更快的大規模查詢（部分數據庫比其餘關係數據庫支持更多的查詢），以及更好的數據壓縮率等。數據入庫後，每每須要其餘方式如數據可視化系統將數據按照規則統計、展示出來，實現數據的監控、指標統計等業務需求，以便充分發揮數據的價值。

業務場景

假設現有一批設備，每一個設備有一個 Client ID，全部設備均經過 MQTT 協議往 MQTT 消息服務器上相應的主題發送數據，主題的設計以下：

devices/{client_id}/messages
複製代碼

每一個設備發送的數據格式爲 JSON，發送的經過該傳感器採集的溫度與溼度數據。

{
    "temperature": 30,
    "humidity" : 20
}
複製代碼

如今須要實時存儲以便在後續任意時間查看數據，提出如下的需求：

• 每一個設備按照每 5 秒鐘一次的頻率進行數據上報，數據庫需存儲每條數據；
• 經過可視化系統查看任意時間區間內的溫度/溼度平均值、最大值、最小值與全部數據中溫度/溼度的平均值。

最終效果

儀表盤右上角能夠選擇時間區間、自動刷新時間，此時設備持續發送數據，儀表盤數據值會隨之變化，實現了比較好的可視化效果。

方案介紹

目前市面上已有多款物聯網消息中間件、時序數據庫和數據可視化產品，結合數據的採集上報、聯網接入、消息存儲與可視化功能來看，EMQ X（高性能物聯網 MQTT 消息中間件） + InfluxDB（時序數據庫）+ Grafana（美觀、強大的可視化監控指標展現工具）組合無疑是最佳的物聯網數據可視化集成方案。

方案總體架構以下圖所示：

• EMQ X：EMQ X 是基於高併發的 Erlang/OTP 語言平臺開發，支持百萬級鏈接和分佈式集羣架構，發佈訂閱模式的開源 MQTT 消息服務器。EMQ X 內置了大量開箱即用的功能，其企業版 EMQ X Enterprise 支持經過規則引擎或消息持久化插件將設備消息高性能地存儲到 InfluxDB，開源用戶需自行處理消息存儲環節。
• InfluxDB：InfluxDB 是一個由 InfluxData 開發的開源時序型數據庫。它由 Go 寫成，着力於高性能地查詢與存儲時序型數據。InfluxDB 被普遍應用於存儲系統的監控數據，IoT 行業的實時數據等場景。
• Grafana： Grafana 是一個跨平臺、開源的度量分析和可視化工具，能夠經過將採集的數據查詢而後可視化的展現。它能夠快速靈活建立的客戶端圖表，面板插件有許多不一樣方式的可視化指標和日誌，官方庫中具備豐富的儀表盤插件，好比熱圖、折線圖、圖表等多種展現方式；支持 Graphite，InfluxDB，OpenTSDB，Prometheus，Elasticsearch，CloudWatch和 KairosDB 等數據源，支持數據項獨立/混合查詢展現；能夠建立自定義告警規則並通知到其餘消息處理服務或組件中。

實現步驟

安裝 EMQ X、InfluxDB 與 Grafana

本文所用各個組件均有 Docker 鏡像，除 EMQ X 須要修改少數配置爲了便於操做使用下載安裝外，InfluxDB 與 Grafana 均使用 Docker 搭建，詳細的安裝步驟本文再也不贅述。

三大部件官網均有不一樣操做系統/平臺的服務或安裝包資源與教程：

• EMQ X：EMQ 官網 emqx.io/cn
• InfluxDB：InfluxData 官網 www.influxdata.com/
• Grafana：Grafana 官網 grafana.com/

EMQ X Enterprise 安裝

安裝

若是您是 EMQ X 新手用戶，推薦經過 EMQ X 指南 快速上手

訪問 EMQ 官網 下載適合您操做系統的安裝包，因爲數據持久化是企業功能，您須要下載 EMQ X 企業版（能夠申請 License 試用） 寫本文的時候 EMQ X 企業版最新版本爲 v3.4.5，該功能須要使用該版本及以上版本，下載 zip 包的啓動步驟以下 ：

## 解壓下載好的安裝包
unzip emqx-ee-macosx-v3.4.4.zip
cd emqx

## 將 License 文件複製到 EMQ X 指定目錄 etc/, License 需自行申請試用或經過購買受權獲取
cp ../emqx.lic ./etc

## 以 console 模式啓動 EMQ X
./bin/emqx console
複製代碼

修改配置

本文中須要用到的配置文件以下：

1. License 文件，EMQ X 企業版 License 文件，使用可用的 License 覆蓋：

etc/emqx.lic
複製代碼

2. EMQ X InfluxDB 消息存儲插件配置文件，用於配置 InfluxDB 鏈接信息、選取入庫主題：

etc/plugins/emqx_backend_influxdb.conf
複製代碼

根據部署實際狀況填寫插件配置信息以下：

backend.influxdb.pool1.server = 127.0.0.1:8089

backend.influxdb.pool1.pool_size = 5

## Whether or not set timestamp when encoding InfluxDB line
backend.influxdb.pool1.set_timestamp = true

## Store Publish Message
## 因爲業務僅需 devices/{client_id}/messages 主題，此處修改默認配置的主題過濾器
backend.influxdb.hook.message.publish.1 = {"topic": "devices/+/messages", "action": {"function": "on_message_publish"}, "pool": "pool1"}
複製代碼

3. EMQ X InfluxDB 消息存儲插件消息模板文件，用於定義消息解析入庫模板：

## 模板文件
data/templates/emqx_backend_influxdb_example.tmpl

## 重命名修改成
data/templates/emqx_backend_influxdb.tmpl
複製代碼

因爲 MQTT Message 沒法直接寫入 InfluxDB, EMQ X 提供了 emqx_backend_influxdb.tmpl 模板文件將 MQTT Message 轉換爲可寫入 InfluxDB 的 DataPoint：

{
  "devices/+/messages": {
    "measurement": "devices",
    "tags": {
      "client_id": "$client_id"
    },
    "fields": {
      "temperature": ["$payload", "temperature"],
      "humidity": ["$payload", "humidity"]
    },
    "timestamp": "$timestamp"
  }
}
複製代碼

關於 EMQ X InfluxDB 使用詳細教程見 InfluxDB 數據存儲

InfluxDB 安裝

經過 Docker 進行安裝，映射數據文件夾與 8089 udp 端口與 8086 端口（Grafana 使用）：

EMQ X 僅支持 InfluxDB UDP 通道，須要 influx_udp 插件支持，且數據庫名稱指定爲 db

## 使用 influx_udp 插件
git clone https://github.com/palkan/influx_udp.git

## 進入插件目錄
cd influx_udp

## 經過插件配置建立並啓動容器
docker run --name=influxdb --rm -d -p 8086:8086 -p 8089:8089/udp \
	-v ${PWD}/files/influxdb.conf:/etc/influxdb/influxdb.conf \
  -e INFLUXDB_DB=db \
  influxdb:latest

## 啓動後檢查容器運行狀態
docker ps -a

複製代碼

至此，能夠重啓 EMQ X 並啓動插件以應用以上配置:

./bin/emqx stop

./bin/emqx start

## 或使用 console 模式能夠看到更多信息
./bin/emqx console

## 啓動插件
./bin/emqx_ctl plugins load emqx_backend_influxdb

## 啓動成功後會有如下提示
Plugin emqx_backend_influxdb loaded successfully.

複製代碼

Grafana 安裝

使用如下命令經過 Docker 安裝並啓動 Grafana：

docker run -d --name=grafana -p 3000:3000 grafana/grafana

複製代碼

啓動成功後瀏覽器訪問 http://127.0.0.1:3000 訪問 Grafana 可視化面板，使用 admin admin 默認用戶名密碼完成初次登陸，登陸後按照提示修改密碼使用新密碼登陸進入主界面：

寫入模擬數據

進行可視化配置以前須要寫入模擬數據，方便配置過程當中進行效果預覽。

如下腳本模擬完成了 100 個設備在過去 12 小時內、每隔 5 秒鐘上報一條模擬溫溼度數據併發送到 EMQ X 的場景，讀者安裝 Node.js 平臺後能夠經過如下命令啓動：

npm install mqtt mockjs --save
node mock.js

複製代碼

模擬腳本執行完畢後，數據將寫入 InfluxDB db 數據庫中，經過如下命令進入 InfluxDB 容器並查看數據：

## 進入 docker 容器
docker exec -it influxdb bash

## 進入 influxdb 命令行
root@581bde65650d:/# influx

## 切換到 db 數據庫
use db;

## 查詢數據
select * from devices limit 1;

## 查詢結果
name: devices
time                client_id      humidity temperature
----                ---------      -------- -----------
1574578725608000000 mock_client_1  54.33    98.5

複製代碼

附：模擬腳本以下：

// Node.js
// mock.js
const mqtt = require('mqtt')
const Mock = require('mockjs')

class MockData {
  constructor(clientNum = 20) {
    this.EMQX_SERVER = 'mqtt://localhost:1883'
    this.clientNum = clientNum
    this.clients = {}
    this.startMock()
  }

  async startMock() {
    const now = Date.now()
    // last 12h every 5s
    for (let ts = now - 12 * 3600 * 1000; ts <= now; ts += 5 * 1000) {
      for (let i = 0; i < this.clientNum; i++) {
        const clientId = `mock_client_${i}`
        const client = this.clients[clientId] || await this.createClient(clientId)
        this.clients[clientId] = client
        const mockData = this.getMockData()
        client.publish(`devices/${clientId}/messages`, JSON.stringify(mockData))
        console.log(`${clientId} send temperature ${mockData.temperature} humidity ${mockData.humidity}`)
      }
    }
  }

  /** * Init a virtual mqtt client * @param {string} clientId ClientID */
  createClient(clientId) {
    return new Promise((resolve, reject) => {
      const client = mqtt.connect(this.EMQX_SERVER, {
        clientId,
      })
      client.on('connect', () => {
        console.log('client s% connected', clientId)
        resolve(client)
      })
      client.on('error', (e) => {
        reject(e)
      })
    })
  }

  /** * Generate mock data */
  getMockData() {
    return {
      temperature: parseFloat(Mock.Random.float(22, 100).toFixed(2)),
      humidity: parseFloat(Mock.Random.float(12, 86).toFixed(2)),
    }
  }
}

// startup
new MockData(100)

複製代碼

可視化配置

組件安裝完成，模擬數據寫入成功後，按照 Grafana 可視化界面的操做指引，完成業務所需數據可視化配置。

添加數據源(Add data source)

添加數據源，即顯示的數據源信息。選取 InfluxDB 類型數據源，輸入鏈接參數進行配置，默認狀況下，關鍵配置信息以下：

• URL：填寫 InfluxDB 鏈接地址，因爲咱們使用 Docker 安裝，Grafana 因爲 InfluxDB 容器網絡不互通，此處能夠輸入當前服務器內網/局域網地址而非 127.0.0.1 或 localhost；
• Auth：InfluxDB 默認啓動無認證方式，根據實際狀況填寫；
• Database：填寫 db ，爲 EMQ X 默認寫入數據庫名。

添加儀表盤(New Dashboard)

添加好數據源後，添加須要顯示的數據儀表盤信息。儀表盤爲多個可視化面板的集合，點擊 New Dashboard 後，選擇 Add Query 經過查詢來添加數據面板：

建立面板須要四個步驟，分別是 Queries(查詢)、Visualization(可視化)、General(圖表配置)、Alert(告警)，下面按照業務需求與完成如下建立流程：

溫、溼度平均值面板

使用 Grafana 的可視化查詢構建工具，查詢出全部設備的平均值：

• FROM：選取數據的 measurement，按照 emqx_backend_influxdb.tmpl 文件配置，此處 measurement 爲 devices；
• SELECT：選取、計算的字段，此處兩個查詢須要使用 Aggregation 功能處理，分別選擇 temperature mean 和 humidity mean，查詢並計算溫度、溼度字段的平均值；
• GROUP BY：默認使用時間區間聚合。
  • time($__interval) 函數表示取 $__interval 時間區間內的數據，如 time(5s) 表示從每 5 秒時間區間原始數據內取出值來進行計算（SELECT 中的計算）
  • fill 參數表示沒有值時候的默認值，爲 null 的時候該數據點不會在圖表顯示出來；
  • tag 可選，按照指定 tag 進行顯示。
• ALIAS BY：該查詢的別名，方即可視化查看。

Visualization 默認不作更改，General 裏面修改面板名稱爲 Device temperature and humidity mean value，若是須要對業務進行監控告警，能夠在 Alert 裏編排告警規則，此處僅作可視化展現，不使用此功能。

完成建立後，點擊左上角返回按鈕，該 Dashboard 裏成功添加一個數據面板。點擊頂部導航欄保存圖標，輸入 Dashboard 名稱完成 Dashboard 的建立。

溫、溼度最大、最小值面板

繼續點擊 Dashboard 的 Add panel 按鈕，添加溫度最大值、最小值圖表。操做步驟同添加平均值，僅對查詢中 SELECT 統計方法字段作出調整，調整爲 Selectors 功能中的 max 和 min 方法。

溫、溼度總平均值、數據條數面板

繼續點擊 Dashboard 的 Add panel 按鈕，添加溫、溼度總平均值，數據條數面板。操做步驟近似上面兩個步驟，分別使用 count 和 mean 方法對指定字段操做，取消 GROUP BY 字段便可完成查詢。Visualization 配置中選擇圖表類型爲 Gauge(儀表) 便可。

保存儀表盤，拖拽調整每一個數據面板大小、位置，最終獲得一個視覺效果較好的數據儀表盤。最終報表完成後，呈現的就是文章開頭展現的效果。

總結

至此咱們完成了 EMQ X 與 InfluxDB + Grafana 物聯網數據可視化集成方案的搭建實現。經過本文，讀者能夠了解到利用 EMQ X 豐富的拓展能力在數據可視化解決方案裏能夠很是快速、靈活地開發出基於 InfluxDB + Grafana 的可視化系統，實現海量數據存儲、計算分析與展示。深刻學習掌握 Grafana 的其餘功能後，用戶能夠定製出更完善的數據可視化乃至監控告警系統。

---

更多信息請訪問咱們的官網 emqx.io，或關注咱們的開源項目 github.com/emqx/emqx ，詳細文檔請訪問 官方文檔。