IoTセンサーデータ基盤の構築とMQTT

4 日前

読了時間: 7分

MQTTは、OASISによるオープンな標準プロトコルです（ISO/IEC 20922）。

概要

MQTTの初期版は1999年に、IBMのAndy Stanford-Clark氏とArlen Nipper氏によって設計され、SCADAによる石油パイプラインの監視に使われました。当時のシステムは異種機器と独自プロトコルが混在し、連携が難しいうえ、高価な衛星回線を使う必要がありました。そのため、帯域を効率よく使い、通信を軽くし、電池の消耗を抑えることが設計の目標でした。

名称の「MQ」は、歴史的にはIBMのMQ製品群（当時のMQSeries）に由来します。ただしMQTTが定義するのはパブリッシュ／サブスクライブ型のメッセージングであり、名前に連想されるキューそのものではありません。IBM公開の3.1仕様ではMQ Telemetry Transportの略と説明されていましたが、OASIS版では単にMQTTという名称が使われています。

技術委員会名にはMessage Queuing Telemetry Transportが含まれますが、2013年の会議では、MQTTを何かの略語として扱わないことが確認されました。OASISの会議議事録を参照できます。

IBMが2013年に仕様をOASISへ提出した後、2014年10月29日にMQTT 3.1.1が公開されました。機能を拡張したMQTT 5.0は2019年3月7日に公開されています。MQTT-SNは、Zigbeeなどの非TCP/IPネットワーク上で動く、電池駆動のセンサーネットワーク向けの派生プロトコルです。

MQTTの特徴

1. コネクション指向：クライアントはサーバーへTCP/IP接続を確立し、明示的な切断やネットワーク障害まで、その接続を使ってメッセージを送受信します。

2. パブリッシュ／サブスクライブ：トピックを基に発行・購読します。クライアント同士が直接やり取りするのではなく、サーバー（ブローカー）が仲介します。

3. QoS（Quality of Service）：発行時に、0、1、2の配信レベルを指定できます。

・QoS 0：最大1回。到達を保証しません。

・QoS 1：少なくとも1回。再送により、重複する場合があります。

・QoS 2：MQTTの送受信区間で、重複を防いで1回の配信を実現します。

4. 制御パケットの種類：代表的なものは次のとおりです。

・CONNECT、CONNACK

・DISCONNECT

・PUBLISH、PUBACK、PUBREC、PUBREL、PUBCOMP

・SUBSCRIBE、SUBACK

・UNSUBSCRIBE、UNSUBACK

・PINGREQ、PINGRESP

MQTTは通信の付加情報が小さく、QoSと柔軟なトピック構造を備えています。企業向けアプリケーションからゲーム、エンターテインメントまで幅広く使われ、特にIoTではテレメトリー通信の代表的な方式になっています。

MQTTパケットの構造

パケットは、最低2バイトの固定ヘッダーと、パケットの種類に応じた可変ヘッダー・ペイロードで構成されます。「固定ヘッダー」という名称でも、Remaining Lengthの符号化により全長が変わる点に注意してください。

最初の4ビットがパケットの種類を表します。

続く下位4ビットは制御フラグです。PUBLISH以外では、仕様で定められた値を使います。

基本的なシーケンスを図に示します。発行者と購読者がCONNECTを送り、ブローカーはCONNACKで応答します。購読者はトピックにSUBSCRIBEし、SUBACKを受け取ります。発行者からブローカーへ、さらに購読者へ送るのがPUBLISHです。QoS 1では受信側がPUBACKで確認します。

CONNECTには、次のようなフィールドが含まれます。

CONNECTに対する応答はCONNACKです。応答コードの意味を次に示します。

メッセージの送受信に使うPUBLISHでは、次のようなフィールドを設定します。

MQTTとIoTアプリケーション

典型的なIoT構成を単純化すると、次のようになります。

センサーが収集した値を、決められたトピックへJSONやCSVで送ります。そのトピックを購読する収集アプリケーションが受け取り、データベースへ保存します。業務アプリケーションは必要に応じて検索・処理します。

なじみ深い構成ですが、さまざまなIoTプロジェクトを経験する中で、主に次の課題が見えてきました。

課題1：MQTTブローカーの可用性

高品質なブローカーは、オープンソースを含めて多数あります。しかし、すべての通信がブローカーを通る構成では、その可用性がサービス全体を制約します。選んだブローカーの開発・運用ノウハウを蓄えることが重要になり、アプリケーション自体を安定化するのと同じくらいの労力が必要になる場合があります。補助的だと思っていた部品が、運用上の中核になるのです。

課題2：重複する保存領域の管理

配信を確実にする構成では、ブローカーがメッセージを保持し、送信先へ渡すストア・アンド・フォワード方式が使われます。収集アプリケーションは、受け取った内容を業務に適した形でデータベースへ保存します。その結果、ブローカー側とアプリケーション側の両方の保存領域を管理し、障害へ備える必要があります。新しいセンサーや形式を追加するたびに、収集アプリケーションの変更も必要になりがちです。

IoT開発者は、常にブローカー運用と二重の保存管理を背負う必要があるのでしょうか。もっと効率よく大量のIoTデータを扱えないか、という問いが生まれます。

MachbaseとMachbase Neo

センサーがMQTTでデータベースへ直接送信できたら、どうでしょうか。

従来のアーキテクチャ：

Machbaseが考えるMQTTインターフェースの活用：

MQTTで直接取り込めるデータベースなら、別のブローカーや重複した保存管理を簡素化でき、専用の収集アプリケーションが不要になる場合もあります。センサーは従来どおりMQTTで送信し、業務側はHTTP API、JDBC、ODBC、Goドライバーなどで検索できます。

この考え方を形にしたのがMachbase Neoです。公式ドキュメントをご覧ください。

Machbaseは、TPCx-IoTの公開性能評価で優れた結果を持つ、韓国発のデータベースです。Neoはそのエンジンに、MQTTサーバーなどIoT向けの機能を統合しています。単一の実行ファイルで配布され、導入が簡単で、IoT開発者の作業の流れに合わせた操作環境を備えています。

ここでは、MQTTからセンサー値を保存する方法を簡単に試します。詳細は公式ドキュメントとチュートリアルを参照してください。デモには mosquitto_pub を使うので、Mosquittoのクライアントツールを用意します。Neoは公式サイト、または次の公式インストールコマンドから入手できます。Windowsではリリースページからダウンロードしてください。

$ sh -c "$(curl -fsSL https://docs.machbase.com/install.sh)"

配布ファイルを展開し、machbase-neo を配置して、serve コマンドで起動します。図の例では、MQTTは5653、HTTPは5654ポートで待ち受けます。

デモ用のタグテーブルを作成します。

$ curl -o - http://127.0.0.1:5654/db/query \
    --data-urlencode \
    "q=create tag table EXAMPLE (name varchar(40) primary key, time datetime basetime, value double)"

mosquitto_pub で、センサーからの送信を再現します。トピックにはテーブル名EXAMPLEが含まれています。

$ mosquitto_pub -h 127.0.0.1 -p 5653 \
    -t db/append/EXAMPLE                   \
    -m '[ "my-car", 1670380342000000000, 32.1 ]'

保存したデータはHTTPで検索できます。

$ curl -o - http://127.0.0.1:5654/db/query \
    --data-urlencode "q=select * from EXAMPLE"

プログラミング例

PythonのMQTTクライアントでデータを送信・保存し、JavaScriptを含むHTMLで可視化してみましょう。

必要なMQTTクライアントライブラリをインストールします。

$ pip install paho-mqtt

エディターで、以下のPythonクライアントを作成します。実行時のファイル名はコード例に合わせてください。

import paho.mqtt.client as mqtt
import time
import random
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("CONNACK OK")
    else:
        print("CONNACK KO code:", rc)
def on_publish(client, userdata, mid):
    print("PUBACK mid:",mid)
mqttClient = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, "python_pub", clean_session=True)
mqttClient.on_connect = on_connect
mqttClient.on_publish = on_publish
mqttClient.connect("127.0.0.1", port=5653, keepalive=10)
mqttClient.loop_start()
for i in range(1000):
    tick = int(time.time())*1000000000
    val  = random.randint(1,9)
    mqttClient.publish("db/append/example", f'["temperature",{tick},{val}]', qos=1)
    time.sleep(1)
mqttClient.disconnect()
mqttClient.loop_stop()

この例は、Neoへ接続して1秒ごとに1〜9の乱数を生成し、temperatureというタグ名と現在時刻を付けて、db/append/example トピックへ発行します。

作成したPythonスクリプトを実行し、NeoのMQTTインターフェースへ送信します。

$ python ./client.py

SQLで、保存された値を確認できます。

次に、データを表示するJavaScriptを含むHTMLを作成します。

<html>
<head>
    <script>
        function loadJS(url) {
            var scriptElement = document.getElementById("chartScript")
            if ( scriptElement != null) {
                scriptElement.remove()
            }
            scriptElement = document.createElement('script');
            scriptElement.id = "chartScript"
            scriptElement.src = url;
            document.getElementsByTagName('head')[0].appendChild(scriptElement);
        }
        function chartData() {
            fetch(`http://127.0.0.1:5654/db/tql`, {
                method: "POST",
                body: `SQL_SELECT("time", "value", from("example", "temperature"), between("last-60s", "last"))
                        MAPVALUE(0, list(value(0), value(1)))
                        POPVALUE(1)
                        CHART(
                            size("600px", "400px"),
                            chartID('rspChart'),
                            theme('dark'),
                            chartOption({
                                xAxis:{ type: "time", axisLabel: { rotate: 30, interval:5 } },
                                yAxis:{ },
                                animation: true,
                                series:[ { type:"line", data:column(0) } ]
                            })
                        )`
            }).then(function(rsp){
                return rsp.json()
            }).then(function(obj){
                document.getElementById('rspChart').style.width = obj.style.width
                document.getElementById('rspChart').style.height = obj.style.height
                obj.jsCodeAssets.forEach(js => loadJS(js))
            })
        }
    </script>
    <script src="http://127.0.0.1:5654/web/echarts/echarts.min.js"></script>
</head>
<body>
    <div id=rspChart></div>
    <script>
        var timer
        function doTimer(){
            timer = setInterval(function(){
                chartData();
            }, 1000)
        }
        doTimer()
    </script>
</body>
</html>

HTMLを保存してブラウザーで開くと、図のようなリアルタイムチャートになります。ブラウザーからNeoへ接続できる環境で実行してください。