EMQX Enterprise

日本語

简体中文
English

日本語

简体中文
English

Appearance

お問い合わせ
お問い合わせ

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データ取り込みは以下のように動作します：

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

特長と利点 ​

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

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

はじめる前に ​

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

前提条件 ​

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

QuasarDBコネクターはODBCを使ってデータベースに接続します。EMQXが稼働するホストにQuasarDB ODBCドライバーをインストールし設定する必要があります。

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

  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

  2. /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

  3. /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インスタンスを起動する方法を説明します。

  1. 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）を使用してください。ホスト名ベースの接続はサポートされていません。

  2. 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を接続するコネクターの作成方法を示します。

  1. EMQXダッシュボードで Integration -> Connectors をクリックします。

  2. 画面右上の Create をクリックします。

  3. Create Connector ページで QuasarDB を選択し、 Next をクリックします。

  4. コネクター名を入力します。大文字・小文字の英数字の組み合わせで、例として my_quasardb とします。

  5. 接続情報を設定します：

    • Server URI：QuasarDBクラスタのURIをIPアドレス形式で入力します。例：qdb://127.0.0.1:2836
    • ODBC Data Source Name：/etc/odbc.iniで定義したDSN名を入力します。例：qdb
    • Username：ユーザー名（あれば）
    • Password：ユーザーのシークレットキー（あれば）
    • Cluster Public Key：クラスタ公開鍵（あれば）

  6. 詳細設定（任意）：高度な設定を参照してください。

  7. Createをクリックする前に、Test ConnectivityをクリックしてEMQXがQuasarDBに接続できるか確認できます。

  8. Createボタンをクリックしてコネクターの設定を完了します。作成成功ダイアログが表示され、ルールを今すぐ作成するか尋ねられます。Create Ruleをクリックすると、コネクターが事前選択された状態でルール作成画面に進みます。Back To Connector Listをクリックすると戻って後でルールを作成できます。

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

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

  1. 前のステップでCreate Ruleをクリックした場合、Add Actionパネルが自動で開き、Type of ActionがQuasarDBに設定され、コネクターが事前選択されています。ステップ5へ進んでください。

    そうでない場合は、EMQXダッシュボードで Integration -> Rules をクリックし、右上の Create をクリック、続いて + Add Action をクリックします。

  2. 左側のSQL EditorにルールIDと以下のSQLを入力し、トピックt/#のメッセージをマッチさせます：

    注意：独自のSQL構文を指定する場合は、Sinkが必要とするすべてのフィールドがSELECT部分に含まれていることを確認してください。

    sql
    SELECT
  *
FROM
  "t/#"

    TIP

    初心者の方は、SQL Examplesをクリックし、Enable Testを有効にしてSQLルールを学習・テストしてください。

  3. 右側のAdd Actionパネルで、Type of ActionドロップダウンからQuasarDBを選択します。ActionドロップダウンはデフォルトのCreate Actionのままにします。

  4. Connectorsドロップダウンから、先ほど作成したmy_quasardbコネクターを選択します。新しいコネクターを作成する場合は、ドロップダウン横のボタンをクリックしてください。設定パラメータはコネクターの作成を参照してください。

  5. Sinkの名前と任意で説明を入力します。

  6. QuasarDBへの書き込み方法を定義するSQL Templateを設定します。

    注意

    SQL Templateは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})

  7. フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。詳細はフォールバックアクションを参照してください。

  8. 詳細設定（任意）：詳細はSinkの高度な設定を参照してください。

  9. Createをクリックする前に、Test ConnectivityをクリックしてSinkがQuasarDBに接続できるかテストできます。

  10. CreateボタンをクリックしてSinkの設定を完了します。新しいSinkがAction Outputsに追加されます。

  11. 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 TimeoutQuasarDBへの接続確立時の最大待機時間5 秒
Start Timeout自動起動リソースが正常になるまでの最大待機時間5 秒
Health Check IntervalQuasarDB接続の自動ヘルスチェック実行間隔15 秒
Health Check Timeout各ヘルスチェックの最大許容時間60 秒

Sinkの高度な設定 ​

フィールド名説明デフォルト値
Buffer Pool SizeEMQXとQuasarDB間のデータフローを処理するバッファワーカープロセス数。高負荷時のスループット向上に増加推奨。16
Request TTLバッファ内でリクエストが有効な最大時間。期限切れのリクエストはキュー内・未アック問わず破棄されます。45 秒
Health Check IntervalQuasarDB接続の自動ヘルスチェック実行間隔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 ModeasyncはQuasarDBの書き込み確認を待たずにパブリッシュを継続し、高スループットを実現。syncは確認を待ってから処理を進めます。非同期モードは順序ずれが発生する可能性があります。Async
Inflight Window未アックのリクエストを同時に許容する最大数。Query Modeがasyncの場合、クライアント単位のメッセージ順序保証のため1に設定推奨。100