OpenTSDB 是可擴充套件的分散式時序資料庫，底層依賴 HBase 並充分發揮了HBase的分散式列儲存特性，支援數百萬每秒的讀寫。

面對大規模快速增長的物聯網感測器採集、交易記錄等資料，時間序列資料累計速度非常快，時序資料庫通過提高效率來處理這種大規模資料，並帶來效能的提升，包括：更高的容納率（Ingest Rates）、更快的大規模查詢（儘管有一些比其他資料庫支援更多的查詢）以及更好的資料壓縮。

本文以 CentOS 7.2 系統中的實際例子來說明如何通過 OpenTSDB 來儲存相關的資訊。

安裝與驗證 OpenTSDB 伺服器

讀者可以參考 OpenTSDB 官方文件 或 Docker

來下載安裝 OpenTSDB 伺服器，本文使用 OpenTSDB 2.4.0 版本。

配置 EMQ X 伺服器

通過 RPM 方式安裝的 EMQ X，OpenTSDB 相關的配置檔案位於 /etc/emqx/plugins/emqx_backend_opentsdb.conf，考慮到功能定位，OpenTSDB 外掛僅支援訊息儲存功能。更多 backend 外掛詳見 EMQ X 資料持久化。

配置連線地址與連線池大小、batch 策略：

## OpenTSDB Server 接入地址
backend.opentsdb.pool1.server = 127.0.0.1:4242

## 連線池大小
backend.opentsdb.pool1.pool_size = 8


## Max batch size of put 最大批量寫條數
backend.opentsdb.pool1.max_batch_size = 20

## 通過 topic 過濾器儲存全部訊息
backend.opentsdb.hook.message.publish.1 = {"topic": "#", "action": {"function": "on_message_publish"}, "pool": "pool1"}
 

**OpenTSDB Backend 訊息儲存規則引數: **

通過 topic 過濾器，設定需要儲存訊息的主題，pool 引數區別多個數據源：

## Store Publish Message
backend.opentsdb.hook.message.publish.1 = {"topic": "#", "action": {"function": "on_message_publish"}, "pool": "pool1"}

啟動該外掛：

./bin/emqx_ctl plugins load emqx_backend_opentsdb
 

訊息模板

由於 MQTT Message 無法直接寫入 OpenTSDB, OpenTSDB Backend 提供了 emqx_backend_opentsdb.tmpl 模板檔案將 MQTT Message 轉換為可寫入 OpenTSDB 的 DataPoint。

訊息模板功能需要重啟 EMQ X 才能應用更改。

tmpl 檔案位於 data/templates/emqx_backend_opentsdb_example.tmpl，使用 json 格式, 使用者可以為不同 Topic 定義不同的 Template, 類似:

{
    "sample": {
        "measurement": "$topic",
        "tags": {
            "host": ["$payload", "data", "$0", "host"],
            "region": ["$payload", "data", "$0", "region"],
            "qos": "$qos",
            "from": "$from"
        },
        "value": ["$payload", "data", "$0", "temp"],
        "timestamp": "$timestamp"
    }
}

其中, measurement 與 fields 為必選項, tags 與 timestamp 為可選項。<Where is value of> 支援通過佔位符如 $key 提取變數名為 key 的變數，支援的變數如下：

• qos: 訊息 QoS
• form: 釋出者資訊
• topic: 釋出主題
• timestamp: 時間戳
• payload.*: JSON 訊息體內任意變數，如 { "data": [{ "temp": 1 }] } 使用 ["$payload", "data", "temp"] 可以提取出 1 來

本示例設定模板如下：

{
    "sample": {
        "measurement": "$topic",
        "tags": {
            "host": ["$payload", "data", "$0", "host"],
            "region": ["$payload", "data", "$0", "region"],
            "qos": "$qos",
            "from": "$from"
        },
        "value": ["$payload", "data", "$0", "temp"],
        "timestamp": "$timestamp"
    }
}

當 Topic 為”sample” 的 MQTT Message 擁有以下 Payload 時:

{
  "data": [
    {
      "temp": 1,
      "host": "serverA",
      "region": "hangzhou"
    },
    {
      "temp": 2,
      "host": "serverB",
      "region": "ningbo"
    }
  ]
}

Backend 會將 MQTT Message 轉換為:

[
  {
    "measurement": "sample",
    "tags": {
      "from": "mqttjs_ebcc36079a",
      "host": "serverA",
      "qos": "0",
      "region": "hangzhou"
    },
    "value": "1",
    "timestamp": "1560743513626681000"
  },
  {
    "measurement": "sample",
    "tags": {
      "from": "mqttjs_ebcc36079a",
      "host": "serverB",
      "qos": "0",
      "region": "ningbo"
    },
    "value": "2",
    "timestamp": "1560743513626681000"
  }
]

使用示例

EMQ X 管理控制檯 WebSocket 頁面中，向 sample 主題釋出如上格式訊息訊息，訊息將解析儲存到 OpenTSDB udp 資料庫對應的 measurement 中。

總結

讀者在理解了 OpenTSDB 中所儲存的資料結構，學習使用訊息模板配置寫入訊息欄位格式後可以結合 OpenTSDB 拓展相關應用。

---

更多資訊請訪問我們的官網 emqx.io，或關注我們的開源專案 github.com/emqx/emqx ，詳細文件請訪問 官方文件。

相關推薦