Apache Doris に MQTT データを取り込む

Apache Doris は、高い同時接続性、高性能、使いやすさで知られる最新のMassively Parallel Processing（MPP）分析データベースシステムです。リアルタイム分析やデータウェアハウジングのシナリオに特に適しています。EMQX 5.10.0 では、MQTT データを Apache Doris と統合でき、効率的なストレージ、リアルタイム分析、強力なデータ可視化を実現します。

本ガイドでは、EMQX と Apache Doris 間のデータ統合の設定および検証方法について実践的な手順を提供します。

注意

EMQX の Apache Doris データ統合は Apache Doris バージョン 2.1.7 以降をサポートしています。

動作の仕組み

Apache Doris データ統合は EMQX の標準機能として提供されており、シンプルな設定で複雑なビジネス開発を可能にします。典型的な IoT アプリケーションでは、EMQX が IoT プラットフォームとしてデバイスの接続とメッセージの送受信を担当し、Apache Doris がデータストレージプラットフォームとしてデバイスの状態やメタデータ、メッセージデータの保存および分析を担当します。

EMQX はルールエンジンと Sink を通じてデバイスのイベントとデータを Apache Doris に転送します。アプリケーションは Apache Doris のデータを読み取り、デバイスの状態を把握したり、デバイスのオンライン・オフライン記録を取得したり、デバイスデータを分析したりできます。具体的なワークフローは以下の通りです：

IoT デバイスが EMQX に接続：IoT デバイスが MQTT プロトコルを介して正常に接続されると、オンラインイベントがトリガーされます。イベントにはデバイスID、送信元IPアドレスなどの属性情報が含まれます。
メッセージのパブリッシュと受信：デバイスは特定のトピックにテレメトリや状態データをパブリッシュします。EMQX はこれらのメッセージを受信すると、ルールエンジン内でマッチング処理を開始します。
ルールエンジンによるメッセージ処理：組み込みのルールエンジンにより、特定のソースからのメッセージやイベントをトピックマッチングに基づいて処理します。ルールエンジンは対応するルールをマッチングし、データ形式の変換、特定情報のフィルタリング、メッセージへのコンテキスト情報の付加などを行います。
Apache Doris への書き込み：ルールがトリガーされるとメッセージを Apache Doris に書き込みます。SQL テンプレートを利用して、ルール処理結果からデータを抽出し SQL を構築して Apache Doris に送信し、メッセージの特定フィールドを対応するテーブルやカラムに書き込んだり更新したりします。

イベントおよびメッセージデータが Apache Doris に書き込まれた後、Apache Doris に接続してデータを読み取り、以下のような柔軟なアプリケーション開発が可能です：

Grafana などの可視化ツールに接続し、データに基づくチャートを生成してデータ変化を表示する。
デバイス管理システムに接続し、デバイス一覧や状態を確認、異常なデバイス挙動を検知し、潜在的な問題を迅速に解消する。

特長と利点

Apache Doris とのデータ統合により、以下の特長と利点をビジネスにもたらします：

柔軟なイベント処理：EMQX ルールエンジンを通じて、Apache Doris はデバイスのライフサイクルイベントを処理でき、IoT アプリケーションの実装に必要な各種管理・監視タスクの開発を大幅に容易にします。イベントデータを分析することで、デバイスの故障や異常挙動、傾向変化を迅速に検知し、適切な対策を講じられます。
メッセージ変換：メッセージは EMQX ルールを通じて広範な処理・変換が可能で、Apache Doris への書き込み前にデータの整形やフィルタリングを行い、保存および活用を容易にします。
リアルタイムデータ取り込み：Apache Doris は HTTP や JDBC インターフェースによるリアルタイムデータ取り込みをサポートします。EMQX と統合することで、MQTT データを低レイテンシで直接 Doris テーブルに書き込め、即時のクエリや分析に適しています。
ストリーミング同期：Apache Doris は Flink、Kafka、トランザクションデータベースなど他のストリーミングデータソースからのリアルタイムデータ取り込みもサポートし、EMQX の MQTT データと組み合わせた統合パイプライン構築に最適です。
標準 SQL とエコシステム互換性：Doris は MySQL 構文に完全対応し標準 SQL をサポートするため、新たな言語を学ぶことなく強力な分析クエリを実行可能です。BI ツールやクライアントアプリケーションとの連携も容易で、ダッシュボード、レポート、自動化ワークフローに活用できます。
ランタイム指標：各 Sink のランタイム指標（総メッセージ数、成功/失敗数、現在の処理レートなど）を閲覧可能です。

柔軟なイベント処理、広範なメッセージ変換、柔軟なデータ操作、リアルタイム監視・分析機能を活用し、効率的で信頼性が高くスケーラブルな IoT アプリケーションを構築し、ビジネスの意思決定や最適化に役立てられます。

はじめる前に

本節では、EMQX ダッシュボードで Apache Doris データ統合を作成する前に必要な準備について説明します。Apache Doris サーバーのインストールやデータテーブルの作成を含みます。

前提条件

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

Apache Doris サーバーのインストール

公式ガイドに従い、Docker Compose を使ってローカルで Doris を起動してください。

データテーブルの作成

MySQL クライアントを使って Doris Frontend に接続し、コマンドを発行します。詳細は公式ドキュメントを参照してください。

例：

mysql -uroot -P9030 -h127.0.0.1

Apache Doris に以下のデータベースと2つのテーブルを作成する必要があります：

emqx_messages テーブル：クライアントID、トピック、ペイロード、作成日時を保存するためのデータテーブル
emqx_client_events テーブル：クライアントID、イベントタイプ、作成日時を保存するためのデータテーブル

sql

create database mqtt;
use mqtt;

create table if not exists
  emqx_messages(
    clientid varchar,
    topic string,
    payload string,
    created_at datetime
  )
  properties (replication_num = 1);

create table if not exists
  emqx_client_events(
    clientid varchar,
    event varchar,
    created_at datetime)
  properties (replication_num = 1);

コネクターの作成

本節では、Sink を Apache Doris サーバーに接続するためのコネクターの作成方法を説明します。

以下の手順は、EMQX と Apache Doris をローカルマシンで実行していることを前提としています。リモートで実行している場合は設定を適宜調整してください。

EMQX ダッシュボードに入り、Integration -> Connectors をクリックします。
ページ右上の Create をクリックします。
Create Connector ページで Doris を選択し、Next をクリックします。
Configuration ステップで以下の情報を設定します：
- Connector name：コネクター名を入力します。大文字・小文字の英数字の組み合わせで、例：my_doris
- Server Host：127.0.0.1:9030 または Apache Doris サーバーがリモートの場合は実際のホスト名を入力
- Database Name：mqtt
- Username：root
- Password：public
高度な設定（任意）：高度な設定を参照してください。
Create をクリックする前に、Test Connectivity をクリックしてコネクターが Apache Doris サーバーに接続できるかテスト可能です。
ページ下部の Create ボタンをクリックしてコネクターの作成を完了します。ポップアップダイアログで Back to Connector List をクリックするか、Create Rule をクリックしてルール作成に進み、Apache Doris へのデータ転送やクライアントイベントの記録を指定できます。詳細はメッセージ保存用 Apache Doris Sink のルール作成およびイベント記録用 Apache Doris Sink のルール作成を参照してください。

メッセージ保存用 Apache Doris Sink のルール作成

本節では、MQTT トピック t/# からのメッセージを処理し、処理済みデータを設定済み Sink を通じて Apache Doris の emqx_messages データテーブルに保存するルールの作成方法を示します。

EMQX と Apache Doris をローカルマシンで実行していることを前提としています。リモートで実行している場合は設定を調整してください。

EMQX ダッシュボードで Integration -> Rules をクリックします。
ページ右上の Create をクリックします。
ルールID に my_rule を入力し、SQL Editor に以下のステートメントを設定します。これはトピック t/# 配下の MQTT メッセージを Apache Doris に保存することを意味します。
注意：独自の SQL 構文を指定する場合は、Sink が必要とするすべてのフィールドを SELECT 部分に含めてください。
sql
```
SELECT
  *
FROM
  "t/#"
```
TIP
初心者の方は SQL Examples と Enable Test をクリックして SQL ルールを学習・テストできます。
- Add Action ボタンをクリックして、ルールによってトリガーされるアクションを定義します。このアクションにより、EMQX はルールで処理したデータを Apache Doris に送信します。
Type of Action ドロップダウンリストから Apache Doris を選択します。Action ドロップダウンはデフォルトの Create Action のままにします。既に作成済みの Sink があれば選択可能です。本デモでは新しい Sink を作成します。
Sink の名前を入力します。名前は大文字・小文字の英数字の組み合わせで指定してください。
Connector ドロップダウンから先ほど作成した my_mysql を選択します。ドロップダウン横のボタンから新しいコネクターを作成することも可能です。設定パラメータはコネクターの作成を参照してください。
使用する機能に応じて SQL Template を設定します：
注意：これは前処理済みの SQL なので、フィールドは引用符で囲まず、文末にセミコロンを付けないでください。
sql
```
INSERT INTO emqx_messages(clientid, topic, payload, created_at) VALUES(
  ${clientid},
  ${topic},
  ${payload},
  FROM_UNIXTIME(${timestamp}/1000)
)
```
SQL テンプレート内に未定義のプレースホルダー変数がある場合、SQL template 上部の Undefined Vars as Null スイッチでルールエンジンの動作を切り替えられます：
- Disabled（デフォルト）：ルールエンジンは文字列 undefined をデータベースに挿入します。
- Enabled：変数が未定義の場合、NULL を挿入することを許可します。
  TIP
  可能な限りこのオプションは有効にしてください。無効にするのは後方互換性を確保する場合のみです。
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。これらはプライマリ Sink がメッセージ処理に失敗した場合にトリガーされます。詳細はフォールバックアクションを参照してください。
高度な設定（任意）：高度な設定を参照してください。
Create ボタンをクリックして Sink の設定を完了します。新しい Sink が Action Outputs に追加されます。
Create Rule ページに戻り、設定内容を確認して Create ボタンをクリックしルールを生成します。

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

また、Integration -> Flow Designer をクリックしてトポロジーを確認すると、トピック t/# 配下のメッセージが Apache Doris に送信・保存されていることがわかります。

イベント記録用 Apache Doris Sink のルール作成

本節では、クライアントのオンライン/オフライン状態を記録し、イベントデータを設定済み Sink を通じて Apache Doris の emqx_client_events テーブルに保存するルールの作成方法を示します。

ルール作成手順はメッセージ保存用 Apache Doris Sink のルール作成とほぼ同様ですが、SQL ルール構文と SQL テンプレートが異なります。

オンライン/オフライン状態記録用のルールは、SQL Editor に以下のステートメントを入力します：

sql

SELECT
  *
FROM
  "$events/client/connected", "$events/client/disconnected"

クライアントイベントデータをデータテーブルに挿入するための SQL テンプレートは以下の通りです：

sql

INSERT INTO emqx_client_events(clientid, event, created_at) VALUES (
  ${clientid},
  ${event},
  FROM_UNIXTIME(${timestamp}/1000)
)

ルールのテスト

MQTTX を使ってトピック t/1 にメッセージを送信し、オンライン/オフラインイベントをトリガーします。

bash

mqttx pub -i emqx_c -t t/1 -m '{ "msg": "hello Apache Doris" }'

2つの Sink の稼働状況を確認すると、新規の受信メッセージ数と送信メッセージ数が1つずつ増加し、イベントレコードが2件追加されているはずです。

emqx_messages データテーブルにデータが書き込まれているか確認します。

bash

mysql> select * from emqx_messages;
+----------+-------+--------------------------+---------------------+
| clientid | topic | payload                  | created_at          |
+----------+-------+--------------------------+---------------------+
| emqx_c   | t/1   | { "msg": "hello Apache Doris" } | 2022-12-09 08:44:07 |
+----------+-------+--------------------------+---------------------+
1 row in set (0.01 sec)

emqx_client_events テーブルにデータが書き込まれているか確認します。

bash

mysql> select * from emqx_client_events;
+----------+---------------------+---------------------+
| clientid | event               | created_at          |
+----------+---------------------+---------------------+
| emqx_c   | client.connected    | 2022-12-09 08:44:07 |
| emqx_c   | client.disconnected | 2022-12-09 08:44:07 |
+----------+---------------------+---------------------+
2 rows in set (0.00 sec)

高度な設定

本節では、Apache Doris コネクターおよび Sink の高度な設定オプションについて詳述します。ダッシュボードでコネクターや Sink を設定する際、Advanced Settings に移動して以下のパラメータをニーズに応じて調整してください。

項目	説明	推奨値
Connection Pool Size	Apache Doris サービスとの接続プール内で維持可能な同時接続数を指定します。この設定により、EMQX と Apache Doris 間のアクティブ接続数を制御し、アプリケーションのスケーラビリティとパフォーマンスを管理します。注意：適切な接続プールサイズはシステムリソース、ネットワークレイテンシ、アプリケーションの負荷などに依存します。大きすぎるとリソース枯渇の恐れがあり、小さすぎるとスループットが制限されます。	`8`
Start Timeout	コネクターが自動起動したリソース（例：Apache Doris のデータベースインスタンス）が正常状態になるまで待機する最大時間（秒）を指定します。この設定により、リソースが完全に稼働しデータ取引可能になるまで処理を進めないようにします。	`5` 秒
Buffer Pool Size	EMQX と Apache Doris 間の egress タイプ Sink でデータフローを管理するバッファワーカーの数を指定します。これらのワーカーはデータを一時的に格納・処理し、ターゲットサービスへの送信を円滑にします。ingress（受信）専用の Sink では適用されないため、"0" に設定可能です。	`16`
Request TTL	バッファに入ったリクエストが有効とみなされる最大時間（秒）を指定します。リクエストはバッファリング開始時からカウントされ、TTL を超えるか、Apache Doris からの応答やアックがタイムリーに得られない場合、期限切れとみなされます。	`45` 秒
Health Check Interval	コネクターが Apache Doris への接続のヘルスチェックを自動的に実施する間隔（秒）を指定します。	`15` 秒
Max Buffer Queue Size	コネクター内の各バッファワーカーがバッファリング可能な最大バイト数を指定します。バッファワーカーは Apache Doris への送信前にデータを一時保管し、データフローの効率化に寄与します。システム性能やデータ転送要件に応じて調整してください。	`256` MB
Max Batch Size	EMQX から Apache Doris へ一度に転送するデータバッチの最大サイズを指定します。サイズを調整することでデータ転送の効率とパフォーマンスを最適化できます。 `1` に設定すると、データレコードはバッチ化せず個別に送信されます。	`1`
Query Mode	メッセージ送信要件に応じて、`asynchronous`（非同期）または `synchronous`（同期）クエリモードを選択できます。非同期モードでは Apache Doris への書き込みが MQTT メッセージのパブリッシュ処理をブロックしませんが、クライアントがメッセージを Apache Doris 到着前に受信する可能性があります。	`Async`
Inflight Window	"インフライトクエリ" とは、開始されたがまだ応答やアックを受け取っていないクエリを指します。コネクターが Apache Doris と通信する際に同時に存在可能なインフライトクエリの最大数を制御します。 Query Mode が `async` の場合、このパラメータは特に重要です。同一 MQTT クライアントからのメッセージを厳密な順序で処理する必要がある場合は、値を 1 に設定してください。	`100`

パスワード認証

スキーマレジストリ

監視

アクセスコントロール

統合

管理

OpenTelemetry との統合

プラグイン

JT/T 808 ゲートウェイ

Apache Doris に MQTT データを取り込む ​

動作の仕組み ​

特長と利点 ​

はじめる前に ​

前提条件 ​

Apache Doris サーバーのインストール ​

データテーブルの作成 ​

コネクターの作成 ​

メッセージ保存用 Apache Doris Sink のルール作成 ​

イベント記録用 Apache Doris Sink のルール作成 ​

ルールのテスト ​

高度な設定 ​

Apache Doris に MQTT データを取り込む

動作の仕組み

特長と利点

はじめる前に

前提条件

Apache Doris サーバーのインストール

データテーブルの作成

コネクターの作成

メッセージ保存用 Apache Doris Sink のルール作成

イベント記録用 Apache Doris Sink のルール作成

ルールのテスト

高度な設定