MQTTとRDBMSのブリッジ

このチュートリアルは、外部のMQTTブローカーからJSONデータを受信し、ブリッジ先のデータベースに保存する手順を説明します。

前提条件

外部のMQTTブローカーには、mosquittoを使用します。 インストールが必要な場合は、公式サイトを参照してください。

この例では、mosquittoサーバーが127.0.0.1:1883で実行中であることを前提とします。

TQLファイルの作成

外部のMQTTブローカーから届くメッセージを処理するTQLファイルを作成します。 この段階のスクリプトは、メッセージペイロードを受け取り、破棄する処理だけを行います。

machbase-neoのWeb UIのファイルエクスプローラーで/mqtt-bridge.tqlを作成し、以下の内容を入力します。

STRING(payload())
DISCARD()

MQTTブリッジの定義

machbase-neoに、MQTTブリッジ「mosquitto」を登録します。

• Name: mosquitto
• Type: MQTT
• Connection String: broker=127.0.0.1:1883 cleansession=true

接続文字列のオプションの詳細は、ドキュメントを参照してください。

「Test」ボタンで接続を確認します。エラーが発生した場合は、接続文字列を修正して再試行します。

MQTTブリッジへのTQLの関連付け

ブリッジの定義とテスト後、mosquittoブローカーの特定のトピックにmqtt-bridge.tqlを関連付けられます。

「Test」ボタンの下の「New subscriber」をクリックし、以下のように設定します。

• Name: mosquitto-sub
• Topic: demo/#
• Destination: 「TQL Script」を選択し、作成したTQLファイルを指定します。

サブスクライバーを作成し、状態を「RUNNING」に設定します。

接続先DBブリッジの定義

外部データベースに接続するブリッジを追加します。この例ではSQLiteを使用しますが、他のデータベースでも、接続文字列以外の手順は同様です。

• Name: destdb
• Type: SQLite
• Connection String file:///tmp/mqtt.db

受信データをmachbase-neo内部だけに保存する場合は、外部データベースのブリッジは不要です。

TQL

mosquittoブリッジがdemo/#トピックでメッセージを受信するたびに実行するTQLコードを作成します。

• 2〜11行目：このJSON文字列はテスト実行に使用します。実際のメッセージがなく、payload()がNULLを返す場合、??演算子が指定したJSON文字列を代わりに使用します。
• 13行目：SCRIPT({}, {})は、JavaScriptを実行するTQL MAP関数です。詳細は、ドキュメントを参照してください。
• 44行目：この例では、SCRIPT MAP関数がすべての処理を行うため、SINKでの追加処理は不要です。TQLはSINK関数で終わる必要があるため、DISCARD()を使用します。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

STRING( payload() ?? `
    {
    "timestamp": 1732653071807,
        "message": {
            "totalCar": "1",
            "reason": "test",
            "total": "1",
            "resetTime": "2024-01-01T23:00:00Z",
            "scenario": "Scenario 0"
        }
    }
`)
SCRIPT({
    // 初期化コードブロック：
    // このブロックは、新しいメッセージの到着ごとに、
    // 最初のレコードがメインブロックに渡る前に1回実行します。
    err = $.db({bridge:"destdb"}).exec("CREATE TABLE IF NOT EXISTS DATA ("+
        "TS INTEGER,"+
        "TOTAL_CAR INTEGER,"+
        "REASON TEXT,"+
        "TOTAL INTEGER,"+
        "RESET_TIME DATETIME)");
    if (err instanceof Error) {
        console.error("Fail to create table", err.message);
    }
}, {
    // メインコードブロック：
    // このブロックは、レコードごとに実行します。
    // この例では、メッセージに1件のレコードだけが含まれます。
    //
    // JSONを解析
    obj = JSON.parse($.values[0]);    
    err = $.db({bridge:"destdb"}).exec("INSERT INTO DATA VALUES(?, ?, ?, ?, ?)",
        obj.timestamp,
        parseInt(obj.message.totalCar),
        obj.message.reason,
        parseInt(obj.message.total),
        obj.message.resetTime,
        obj.message.scenario);
    if (err instanceof Error) {
        console.error("Fail to insert into table", err.message);
    }
})
DISCARD()

データの発行

まず、テスト用のJSONファイルを用意します。

mqtt-test.json

{
  "timestamp": 1732653071807,
  "message": {
     "totalCar": "3683",
     "reason": "car",
     "total": "3956",
     "resetTime": "2024-11-25T23:00:00Z",
     "scenario": "Scenario 1"
  }
}

demo/sensor_1トピックにメッセージを発行します。ペイロードには、作成したファイルの内容を使用します。

mosquitto_pub -h 127.0.0.1 -p 1883 \
  -t demo/sensor_1 \
  -f ./mqtt-test.json

ブリッジ先DBの検索

接続先データベースにデータが保存されたことを確認します。machbase-neoのSQLエディターで-- env: bridge=destdbコメントを使用すると、ブリッジ先DBに直接クエリを実行できます。

-- env: bridge=destdb
SELECT * FROM DATA;
-- env: reset