QuasarDBへのMQTTデータ取り込み

QuasarDBは、大量のタイムスタンプ付きデータの保存とクエリに特化した高性能なカラム指向時系列データベースです。EMQXはQuasarDBとの統合をサポートしており、MQTTメッセージやクライアントイベントをQuasarDBに保存できます。これにより、IoTテレメトリの管理や分析のためのデータパイプラインや分析プロセスの構築が容易になります。

本ページでは、EMQXとQuasarDB間のデータ統合の詳細な概要と、データ統合の作成および検証に関する実践的な手順を提供します。

動作概要

QuasarDBデータ統合はEMQXに標準搭載された機能であり、EMQXのデバイス接続およびメッセージ伝送機能と、QuasarDBの高性能な時系列ストレージを組み合わせています。組み込みのルールエンジンコンポーネントとSinkを通じて、MQTTメッセージやクライアントイベントをQuasarDBに保存できます。この統合により、EMQXからQuasarDBへのデータ取り込みが簡素化され、複雑なコーディングを必要としません。

以下の図は、EMQXとQuasarDB間の典型的なデータ統合アーキテクチャを示しています。

quasardb_integration

QuasarDBへのMQTTデータ取り込みは以下のように動作します：

メッセージのパブリッシュと受信：IoTデバイスはMQTTプロトコルを介してEMQXに正常に接続し、リアルタイムのMQTTデータをパブリッシュします。EMQXがこれらのメッセージを受信すると、ルールエンジン内でマッチング処理を開始します。
メッセージデータの処理：メッセージが到着するとルールエンジンを通過し、EMQXで定義されたルールによって処理されます。ルールは事前定義された条件に基づき、QuasarDBにルーティングすべきメッセージを判別します。ペイロード変換が指定されている場合は、データ形式の変換、特定情報のフィルタリング、追加コンテキストによるペイロードの強化などが適用されます。
QuasarDBへのデータ取り込み：ルールがトリガーされると、メッセージのQuasarDBへの書き込みが開始されます。SQLテンプレートを用いて、ルール処理結果からデータを抽出しSQLを構築、QuasarDBに送信して実行することで、メッセージの特定フィールドを対応するテーブルに書き込みます。
データの保存と活用：データがQuasarDBに保存されることで、企業はその時系列クエリ機能を活用し、分析、監視、運用用途に利用できます。

特長と利点

QuasarDBとのデータ統合は以下の特長と利点を提供します：

リアルタイムデータストリーミング：EMQXはリアルタイムデータストリームの処理に最適化されており、ソースシステムからQuasarDBへの効率的かつ信頼性の高いデータ伝送を実現します。即時の洞察やアクションが必要なユースケースに最適です。
高性能な時系列ストレージ：QuasarDBのカラム型エンジンは時系列ワークロードに最適化されており、高速な取り込みスループットと大量のタイムスタンプ付きデータに対する効率的な範囲クエリを提供します。
柔軟なデータ変換：EMQXは強力なSQLベースのルールエンジンを備え、QuasarDBに保存する前のデータ前処理を可能にします。フィルタリング、ルーティング、集約、強化など多様なデータ変換機構をサポートします。
バッチ処理対応：QuasarDB Sinkはバッチ書き込みをサポートし、往復回数を削減して全体の取り込みスループットを向上させます。

はじめる前に

このセクションでは、QuasarDBデータ統合の作成前に必要な準備、ODBCドライバーの設定やQuasarDBのインストール方法について説明します。

前提条件

EMQXデータ統合のルールに関する知識
データ統合に関する知識

ODBCドライバーのインストールと設定

QuasarDBコネクターはODBCを使ってデータベースに接続します。EMQXが稼働するホストにQuasarDB ODBCドライバーをインストール・設定してからコネクターを作成してください。

完全なインストール手順はQuasarDB ODBCドキュメントを参照してください。以下はDebian系システムでドライバー3.14.1を使った典型的なセットアップ例です。

QuasarDB C APIパッケージとODBCドライバーをダウンロード・インストールします：

bash

curl -fsSL -O https://download.quasar.ai/quasardb/3.14/3.14.1/api/c/qdb-api_3.14.1.deb
curl -fsSL -O https://download.quasar.ai/quasardb/3.14/3.14.1/api/odbc/qdb-3.14.1-linux-64bit-odbc-driver.tar.gz
apt-get install -yqq ./qdb-api_3.14.1.deb
tar -C /tmp/qdb_odbc_driver -xf qdb-3.14.1-linux-64bit-odbc-driver.tar.gz

/etc/odbcinst.iniにドライバーを登録します：

ini

[qdb_odbc_driver]
Description=Quasardb ODBC Driver
Driver=/tmp/qdb_odbc_driver/lib/libqdb_odbc_driver.so
Setup=/tmp/qdb_odbc_driver/lib/libqdb_odbc_driver.so

/etc/odbc.iniにデータソース名（DSN）を作成します：

ini

[qdb]
Driver = qdb_odbc_driver
Description = QuasarDB ODBC Data Source
#URI = qdb://172.100.239.30:2836
#UID = user_name
#PWD = user_key
#KEY = cluster_public_key

ここで設定したDSN名（例：qdb）は、コネクター作成時のODBC Data Source Nameフィールドに入力します。

QuasarDBのインストールと接続

Dockerを使ってQuasarDBインスタンスを起動する方法を説明します。

QuasarDBのDockerイメージをプルして起動します：
bash
```
docker run -d --name qdb \
  -p 2836:2836 \
  bureau14/qdb:3.14.1
```
TIP
QuasarDBはホスト名ではなくIPアドレスでの接続を要求します。URIには127.0.0.1（または実際のホストIP）を使用してください。ホスト名ベースの接続はサポートされていません。
QuasarDBシェルで接続し、インスタンスが起動していることを確認します：
bash
```
docker run -it --rm bureau14/qdbsh --cluster qdb://127.0.0.1:2836
```

ユーザー認証やクラスタキー認証を有効にする場合は、QuasarDBセキュリティドキュメントを参照してください。

テーブルの作成

取り込んだデータを格納するためにQuasarDBでテーブルを作成します。以下は温度と湿度の読み取り値を格納するテーブル作成例です：

sql

CREATE TABLE temp_hum (temp DOUBLE, hum DOUBLE);

TIP

QuasarDBのテーブルには常に暗黙の$timestampインデックス列が含まれます。テーブル作成時に宣言する必要はありませんが、INSERT文で参照可能です。

コネクターの作成

EMQXとQuasarDBを接続するコネクターの作成手順を説明します。

EMQXダッシュボードでIntegration -> Connectorsをクリックします。
画面右上のCreateをクリックします。
Create ConnectorページでQuasarDBを選択し、Nextをクリックします。
コネクター名を入力します。英数字の組み合わせで、例としてmy_quasardbなどが使えます。
接続情報を設定します：
- Server URI：IPアドレスを使ったQuasarDBクラスタのURIを入力します。例：qdb://127.0.0.1:2836
- ODBC Data Source Name：/etc/odbc.iniで定義したDSN名を入力します。例：qdb
- Username：ユーザー名（あれば）
- Password：ユーザーのシークレットキー（あれば）
- Cluster Public Key：クラスタ公開鍵（あれば）
詳細設定（任意）：詳細設定を参照してください。
Createをクリックする前に、Test ConnectivityでEMQXがQuasarDBに接続できるか確認できます。
Createをクリックしてコネクターの作成を完了します。作成成功ダイアログが表示され、ルールを作成するか尋ねられます。Create Ruleをクリックするとコネクターが事前選択された状態でルール作成画面に進みます。後でルールを作成する場合はBack To Connector Listをクリックしてください。

QuasarDB Sinkを使ったルールの作成

ここでは、ソースMQTTトピックt/#からのメッセージを処理し、QuasarDBのtemp_humテーブルに保存するルールの作成方法を説明します。

前のステップでCreate Ruleをクリックした場合、Add Actionパネルが自動で開き、Type of ActionがQuasarDB、コネクターが事前選択されています。ステップ5へ進んでください。
そうでない場合は、EMQXダッシュボードでIntegration -> Rulesをクリックし、右上のCreateをクリック、続けて**+ Add Action**をクリックします。
左側のSQL EditorにルールIDと以下のSQLを入力し、トピックt/#のメッセージにマッチさせます：
※独自のSQL構文を指定する場合は、Sinkが必要とする全てのフィールドがSELECT句に含まれていることを確認してください。
sql
```
SELECT
  *
FROM
  "t/#"
```
TIP
初心者の方はSQL Examplesをクリックし、Enable TestでSQLルールを学習・テストできます。
右側のAdd Actionパネルで、Type of ActionドロップダウンからQuasarDBを選択します。ActionはデフォルトのCreate Actionのままにします。
Connectorsドロップダウンから先ほど作成したmy_quasardbコネクターを選択します。ドロップダウン横のボタンから新規コネクター作成も可能です。設定パラメータはコネクターの作成を参照してください。
Sinkの名前と任意の説明を入力します。
QuasarDBへの書き込み方法を定義するSQL Templateを設定します。
注意
SQLテンプレートはINSERT文のみを受け付けます。UPDATEやDELETEなど他の文はサポートされていません。
SQLテンプレートは${clientid}などのプレースホルダー変数をサポートします。QuasarDBは暗黙のタイムスタンプインデックス列として$timestampを使用し、now()で現在のサーバー時刻を挿入できます。
注意
QuasarDB ODBCドライバーはプリペアドステートメントをサポートしていません。STRINGまたはBLOB型に解決される値は、SQLテンプレート内で手動でシングルクォート（'）で囲む必要があります。
sql
```
insert into temp_hum($timestamp, temp, hum)
values (now(), ${.temp}, ${.hum})
```
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。詳細はフォールバックアクションを参照してください。
詳細設定（任意）：Sink詳細設定を参照してください。
Createをクリックする前に、Test ConnectivityでSinkがQuasarDBに接続できるかテストできます。
CreateをクリックしてSink設定を完了します。新しいSinkがAction Outputsに追加されます。
Create Ruleページに戻り、設定内容を確認してSaveをクリックしルールを生成します。

これでルールの作成が完了しました。Integration -> Rulesページで新規ルールを確認できます。**Actions(Sink)**タブをクリックすると新しいQuasarDB Sinkが表示されます。

また、Integration -> Flow Designerをクリックするとトポロジーを確認でき、トピックt/#のメッセージがルールmy_ruleで処理された後QuasarDBに転送されていることを検証できます。

ルールのテスト

MQTTXを使ってトピックt/1にメッセージを送信し、ルールをトリガーします。

bash

mqttx pub -i emqx_c -t t/1 -m '{ "temp": "27.5", "hum": "41.8" }'

QuasarDB Sinkの稼働統計を確認してください。1件の新規マッチと1件の新規送信メッセージがあるはずです。QuasarDBのtemp_humテーブルにデータが書き込まれていることを検証してください。

詳細設定

QuasarDBコネクターおよびSinkの詳細設定オプションについて説明します。ダッシュボードで設定する際はAdvanced Settingsを展開し、用途に応じて以下のパラメータを調整できます。

コネクター詳細設定

フィールド名	説明	デフォルト値
Connection Pool Size	プール内で維持する同時接続数。大きすぎるとシステム資源を枯渇させ、小さすぎるとスループットが制限されます。	`8`
Connect Timeout	QuasarDBへの接続確立時に待機する最大時間。	`5`秒
Start Timeout	自動起動リソースが正常になるまで待機する最大時間。	`5`秒
Health Check Interval	QuasarDB接続に対して自動ヘルスチェックを実行する間隔。	`15`秒
Health Check Timeout	各ヘルスチェックの最大許容時間。	`60`秒

Sink詳細設定

フィールド名	説明	デフォルト値
Buffer Pool Size	EMQXとQuasarDB間のデータフローを処理するバッファワーカー数。負荷が高い場合は増やしてスループットを向上させます。	`16`
Request TTL	バッファ内でリクエストが有効な最大時間。キュー内または未アックのままTTLを超えたリクエストは破棄されます。	`45`秒
Health Check Interval	QuasarDB接続に対してSinkが自動ヘルスチェックを実行する間隔。	`15`秒
Health Check Interval Jitter	複数ノードが同時にチェックしないようにヘルスチェック間隔にランダム遅延を追加。複数のActionやSourceが同一コネクターを共有する場合に有効。	`0`ミリ秒
Health Check Timeout	各Sinkヘルスチェックの最大許容時間。	`60`秒
Max Buffer Queue Size	各バッファワーカーが保持可能な最大バイト数。バーストが多い場合は増やしてください。	`256`MB
Batch Size	一度にQuasarDBへ送信する最大レコード数。`1`に設定するとバッチ処理を無効化し個別送信になります。	`100`
Query Mode	`async`はQuasarDBの書き込み確認を待たずにパブリッシュを継続。`sync`は確認を待ちます。asyncはスループットが高い反面、順序が前後する可能性があります。	`Async`
Inflight Window	未アックのリクエストを同時に許容する最大数。Query Modeが`async`の場合、クライアント単位のメッセージ順序保証のため`1`に設定してください。	`100`

Kubernetes

EMQX Operator

EMQX クラスター管理

API リファレンス

パスワード認証

監視

アクセスコントロール

統合

管理

Prometheus との統合

OpenTelemetry との統合

エンドツーエンドトレース

プラグインカタログ

JT/T 808 ゲートウェイ

GB/T 32960 ゲートウェイ

Volcengine RTC

シナリオ

GPT-Realtime

QuasarDBへのMQTTデータ取り込み ​

動作概要 ​

特長と利点 ​

はじめる前に ​

前提条件 ​

ODBCドライバーのインストールと設定 ​

QuasarDBのインストールと接続 ​

テーブルの作成 ​

コネクターの作成 ​

QuasarDB Sinkを使ったルールの作成 ​

ルールのテスト ​

詳細設定 ​

コネクター詳細設定 ​

Sink詳細設定 ​