如何使用規則引擎存儲消息到OpenTSDB數據庫

如何使用規則引擎存儲消息到OpenTSDB數據庫，很多新手對此不是很清楚，為了幫助大家解決這個難題，下面小編將為大家詳細講解，有這方面需求的人可以來學習下，希望你能有所收獲。

OpenTSDB 介紹

OpenTSDB 是可擴展的分布式時序數據庫，底層依賴 HBase 并充分發揮了HBase的分布式列存儲特性，支持數百萬每秒的讀寫。

面對大規模快速增長的物聯網傳感器采集、交易記錄等數據，時間序列數據累計速度非常快，時序數據庫通過提高效率來處理這種大規模數據，并帶來性能的提升，包括：更高的容納率（Ingest Rates）、更快的大規模查詢以及更好的數據壓縮。

安裝與驗證 OpenTSDB 服務器

讀者可以參考 OpenTSDB 官方文檔 (http://opentsdb.net) 或 Docker (https://hub.docker.com/r/petergrace/opentsdb-docker/) 來下載安裝 OpenTSDB 服務器，本文使用 OpenTSDB 2.4.0 版本。

場景介紹

該場景需要將 EMQ X 指定主題下且滿足條件的消息存儲到 OpenTSDB 數據庫。為了便于后續分析檢索，消息內容需要進行拆分存儲。

該場景下客戶端上報數據如下：

• Topic：stat/cpu

• Payload:

  {
    "metric": "cpu",
    "tags": {
      "host": "serverA"
    },
    "value":12
  }

  
  

準備工作

啟動 OpenTSDB Server

啟動 OpenTSDB Server 并開放 4242 端口。

$ docker pull petergrace/opentsdb-docker

$ docker run -d --name opentsdb -p 4242:4242 petergrace/opentsdb-docker

配置說明

創建資源

打開 EMQ X Dashboard，進入左側菜單的 資源 頁面，點擊 新建 按鈕，選擇 OpenTSDB 資源類型并完成相關配置進行資源創建。

創建規則

進入左側菜單的 規則 頁面，點擊 新建 按鈕，進行規則創建。這里選擇觸發事件 message.publish，即在 EMQ X 收到 PUBLISH 消息時觸發該規則進行數據處理。

選定觸發事件后，我們可在界面上看到可選字段及示例 SQL：

篩選所需字段

規則引擎使用 SQL 語句過濾和處理數據。例如前文提到的場景中我們需要將 payload 中的字段提取出來使用，則可以通過 payload.<fieldName> 實現。同時我們僅僅期望處理 stat/cpu 主題，那么可以在 WHERE 子句中使用主題通配符 =~ 對 topic 進行篩選：topic =~ 'stat/cpu'， 最終我們得到 SQL 如下：

SELECT
  payload.metric as metric, payload.tags as tags, payload.value as value
FROM
  "message.publish"
WHERE
	topic =~ 'stat/cpu'

SQL 測試

借助 SQL 測試功能，我們可以快速確認剛剛填寫的 SQL 語句能否達成我們的目的。首先填寫用于測試的 payload 等數據如下：

然后點擊 測試 按鈕，我們得到以下數據輸出：

{
  "metric": "cpu",
  "tags": {
    "host": "serverA"
  },
  "value": 12
}

測試輸出與預期相符，我們可以進行后續步驟。

添加響應動作，存儲消息到 OpenTSDB

SQL 條件輸入輸出無誤后，我們繼續添加相應動作，配置寫入 SQL 語句，將篩選結果存儲到 OpenTSDB。

點擊響應動作中的 添加 按鈕，選擇 保存數據到 OpenTSDB 動作，選取剛剛創建的 OpenTSDB 資源并完成剩余參數設置。OpenTSDB 動作用到的幾個參數分別為：

1. 詳細信息。是否需要 OpenTSDB Server 返回存儲失敗的 Data points 及失敗原因，默認為 false。

2. 摘要信息。是否需要 OpenTSDB Server 返回 data point 存儲成功與失敗的數量，默認為 true。

3. 最大批處理數量。消息請求頻繁時允許驅動從隊列中一次讀取多少個 Data Points 合并為一個 HTTP 請求，為性能優化參數，默認為 20。

4. 是否同步調用。配置 OpenTSDB Server 是否等待所有數據都被寫入后才返回結果，默認為 false。

5. 同步調用超時時間。OpenTSDB Server 等待數據寫入的最大時間，默認為 0，即永不超時。

這里我們全部使用默認配置，點擊 新建 按鈕完成規則創建。

測試

預期結果

我們成功創建了一條規則，包含一個處理動作，動作期望效果如下：

1. 客戶端向 stat/cpu 主題上報消息時，該消息將命中 SQL，規則列表中 已命中 數字增加 1；

2. OpenTSDB Server 中將增加一條數據，數據內容與消息內容一致。

使用 Dashboard 中的 Websocket 工具測試

切換到 工具 --> Websocket 頁面，使用任意信息客戶端連接到 EMQ X，連接成功后在 消息 卡片中發送如下消息：

• Topic：stat/cpu

• Payload:

  {
    "metric": "cpu",
    "tags": {
      "host": "serverA"
    },
    "value":12
  }

  
  

點擊 發送 按鈕，發送成功后可以看到當前規則已命中次數已經變為了 1。

然后通過 Postman 向 OpenTSDB 發送查詢請求，當我們得到如下應答時說明新的 data point 已經添加成功：

至此，我們通過規則引擎實現了使用規則引擎存儲消息到 OpenTSDB 數據庫的業務開發。

看完上述內容是否對您有幫助呢？如果還想對相關知識有進一步的了解或閱讀更多相關文章，請關注億速云行業資訊頻道，感謝您對億速云的支持。