|
IoTDB 是最早由清華大學發起的開源時序數據庫項目，現已經是 Apache 的頂級項目。IoTDB 可以為用戶提供數據收集、存儲和分析等服務。由于其輕量級架構、高性能和高可用的特性，以及與 Hadoop 和 Spark 生態的無縫集成，滿足了工業 IoT 領域中海量數據存儲、高吞吐量數據寫入和復雜數據查詢分析的需求。 EMQX 是一個大規模擴展、可彈性伸縮的開源云原生分布式物聯網消息中間件，由開源物聯網數據基礎設施軟件供應商 EMQ 映云科技 發布。EMQX 可以高效可靠地處理海量物聯網設備的并發連接，并且內置了強大的規則引擎功能，用以對事件和消息流數據進行高性能地實時處理。規則引擎通過 SQL 語句提供了靈活的 "配置式" 的業務集成方案，簡化了業務開發流程，提升了易用性，降低了用戶的業務邏輯與 EMQX 的耦合度。 本文將介紹如何使用 EMQX 規則引擎的 MQTT 數據橋接功能，接收 MQTT 客戶端發送的數據，并實時插入到時序數據庫 IoTDB。 準備工作本文示例中用到的軟件和環境： - 操作系統： Mac OSX
- IoTDB： Binary 包（Server），版本 0.12.4
- MQTT 服務器： EMQX 開源版 v4.3.11
- MQTT 客戶端軟件：MQTTX v1.6.0
IoTDB 安裝首先我們需要從 IoTDB 官方頁面下載 IoTDB Server（單機版）的二進制包。 下載完成之后解壓，進入解壓后的目錄： % lsLICENSE README.md RELEASE_NOTES.md data ext licenses sbinNOTICE README_ZH.md conf docs lib logs tools
要啟用 IoTDB 的 MQTT 協議支持，需要改動 IoTDB 的配置文件 conf/iotdb-engine.properties： *后續建模使用了一個存儲組 root.sg，為了增加寫入并行度，需要同時將 iotdb-engine.properties 中的 virtual_storage_group_num 設置為機器核數。
####################### MQTT Broker Configuration##################### whether to enable the mqtt service.enable_mqtt_service=true# the mqtt service binding host.mqtt_host=0.0.0.0# the mqtt service binding port.mqtt_port=2883# the handler pool size for handing the mqtt messages.mqtt_handler_pool_size=1# the mqtt message payload formatter.mqtt_payload_formatter=json# max length of mqtt message in bytemqtt_max_message_size=1048576
其中 enable_mqtt_service 默認為 false，需要改成 true。mqtt_port 默認值是 1883，為了避免與 emqx 的端口號沖突，需要改為 2883。 然后使用 ./sbin/start-server.sh 啟動 IoTDB 服務端： % ./sbin/start-server.sh---------------------Starting IoTDB---------------------Maximum memory allocation pool = 2048MB, initial memory allocation pool = 512MBIf you want to change this configuration, please check conf/iotdb-env.sh(Unix or OS X, if you use Windows, check conf/iotdb-env.bat).2022-01-10 14:15:31,914 [main] INFO o.a.i.d.c.IoTDBDescriptor:121 - Start to read config file file:./sbin/../conf/iotdb-engine.properties...2022-01-10 14:14:28,690 [main] INFO o.a.i.d.s.UpgradeSevice:73 - Upgrade service stopped2022-01-10 14:14:28,690 [main] INFO o.a.i.db.service.IoTDB:153 - Congratulation, IoTDB is set up successfully. Now, enjoy yourself!2022-01-10 14:14:28,690 [main] INFO o.a.i.db.service.IoTDB:101 - IoTDB has started
我們保持這個終端窗口不動，另外打開一個新的命令行終端窗口，啟動 IoTDB 的 shell 工具： % ./sbin/start-cli.sh---------------------Starting IoTDB Cli---------------------_____ _________ ______ ______|_ _| | _ _ ||_ _ `.|_ _ \| | .--.|_/ | | \_| | | `. \ | |_) || | / .'`\ \ | | | | | | | __'._| |_| \__. | _| |_ _| |_.' /_| |__) ||_____|'.__.' |_____| |______.'|_______/ version 0.12.4IoTDB> login successfullyIoTDB>
至此 IoTDB 環境就準備好了。如要了解 IoTDB 的基本使用方法，可以參考官網的快速上手頁面。 安裝、配置 EMQX下載和啟動 EMQX我們直接使用命令行下載 macOS 版本的 EMQX 開源版，更多安裝包請訪問 EMQX 開源版下載頁面。 % wget https://www.emqx.com/en/downloads/broker/4.3.11/emqx-macos-4.3.11-amd64.zip
然后解壓并啟動 EMQX： % unzip -q emqx-macos-4.3.11-amd64.zip% cd emqx% ./bin/emqx consolelog.to = "console"Erlang/OTP 23 [erts-11.1.8] [emqx] [64-bit] [smp:8:8] [ds:8:8:8] [async-threads:4] [hipe]Starting emqx on node emqx@127.0.0.1Start mqtt:tcp:internal listener on 127.0.0.1:11883 successfully.Start mqtt:tcp:external listener on 0.0.0.0:1883 successfully.Start mqtt:ws:external listener on 0.0.0.0:8083 successfully.Start mqtt:ssl:external listener on 0.0.0.0:8883 successfully.Start mqtt:wss:external listener on 0.0.0.0:8084 successfully.Start http:management listener on 8081 successfully.Start http:dashboard listener on 18083 successfully.EMQX Broker 4.3.11 is running now!Eshell V11.1.8 (abort with ^G)(emqx@127.0.0.1)1>
配置規則使用瀏覽器打開 EMQX Dashboard，在規則引擎頁面創建一條規則： SQL 語句為： SELECT clientid, now_timestamp('millisecond') as now_ts_ms, payload.bar as barFROM "t/#"
然后我們在頁面的底部，給規則加一個 "橋接數據到 MQTT Broker" 動作： 這個動作需要關聯一個資源，我們點擊右上角的 “新建資源” 來創建一個 MQTT Bridge 資源： 遠程 Broker 地址要填寫 IoTDB 的 MQTT 服務地址，即 "127.0.0.1:2883"。客戶端 Id、用戶名、密碼都填寫 root，因為 root 是 IoTDB 默認的用戶名和密碼。 其他選項保持默認值不變，點擊 ”測試連接“ 按鈕確保配置無誤，然后再點擊右下角的 ”新建“ 按鈕創建資源。 現在返回到動作創建頁面，關聯資源的下拉框里自動填充了我們剛才創建的資源。 現在我們繼續填寫更多的動作參數： IoTDB 不關心消息主題，我們填一個任意的主題：foo。 IoTDB 要求消息內容是一個 JSON 格式，消息內容模板可以按照上圖中樣式填寫。詳情請參見 IoTDB 的通信服務協議文檔。 { "device": "root.sg.${clientid}", "timestamp": ${now_ts_ms}, "measurements": [ "bar" ], "values": [ ${bar} ]}
注意其中的 "${clientid}", "${now_ts_ms}" 以及 "${bar}" 都是從規則的 SQL 語句的輸出中提取的變量，所以必須保證這些變量跟 SQL 語句的 SELECT 字句對應上。 現在可以點擊 ”確認“ 保存動作配置，然后再次點擊 ”新建“ 完成規則的創建。 使用 MQTT Client 發送消息接下來我們使用 MQTT 客戶端工具 - MQTT X，來發送一條消息給 EMQX： MQTT X 是 EMQ 發布的一款完全開源的 MQTT 5.0 跨平臺桌面客戶端。支持快速創建多個同時在線的 MQTT 客戶端連接，方便測試 MQTT/TCP、MQTT/TLS、MQTT/WebSocket 的連接、發布、訂閱功能及其他 MQTT 協議特性。
MQTT 客戶端的連接參數里面，我們只需要填一個參數，Client ID："abc"，其他的保持默認值不變。 連接成功之后，我們發送 2 條主題為："t/1" 的消息，消息內容格式為： { "bar": 0.2}
然后回到 EMQX Dashboard 的規則引擎頁面，觀察規則的命中次數，確認規則被觸發了 2 次： 最后我們回到命令行終端的 IoTDB 客戶端窗口，使用下面的 SQL 語句查詢數據： IoTDB> SHOW TIMESERIES root.sg.abc+---------------+-----+-------------+--------+--------+-----------+----+----------+| timeseries|alias|storage group|dataType|encoding|compression|tags|attributes|+---------------+-----+-------------+--------+--------+-----------+----+----------+|root.sg.abc.bar| null| root.sg| FLOAT| GORILLA| SNAPPY|null| null|+---------------+-----+-------------+--------+--------+-----------+----+----------+Total line number = 1It costs 0.006sIoTDB> SELECT * FROM root.sg.abc+-----------------------------+---------------+| Time|root.sg.abc.bar|+-----------------------------+---------------+|2022-01-10T17:39:41.724+08:00| 0.3||2022-01-10T17:40:32.805+08:00| 0.2|+-----------------------------+---------------+Total line number = 2It costs 0.007sIoTDB>
數據插入成功！ 結語至此，我們完成了通過 EMQX 規則引擎功能將消息持久化到 IoTDB 時序數據庫。 在實際生產場景中，我們可以使用 EMQX 處理海量的物聯網設備并發連接，并通過規則引擎靈活地處理業務功能，然后將設備發送的消息持久化到 IoTDB 數據庫，最后使用 Hadoop/Spark、Flink 或 Grafana 等對接 IoTDB 實現大數據分析、可視化展示等。 EMQX + IoTDB 的組合是一個簡潔、高效且易擴展、高可用的服務端集成方案，對于物聯網設備管理和數據處理場景來說，是一個不錯的選擇。
原創文章，作者：EMQ，如若轉載，請注明出處：https://www.emqx.com/zh/blog/store-mqtt-messages-to-time-series-database-iotdb
| 
發表于 2022-7-22 15:16:58
收藏
頂
踩