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

Apache Doris は、高い同時接続性、高性能、使いやすさで知られる最新の MPP（Massively Parallel Processing）分析データベースシステムです。リアルタイム分析やデータウェアハウスのシナリオに特に適しています。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 には以下のデータベースとテーブルを作成する必要があります：

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 に転送するデータやクライアントイベントを記録する Sink を指定するルール作成を続けられます。詳細はメッセージ保存用 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 スイッチでルールエンジンの動作を切り替えられます：
- 無効（デフォルト）：ルールエンジンは未定義変数に文字列 undefined を挿入します。
- 有効：未定義変数に対して 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 エディターには以下のステートメントを入力します：

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`

Kubernetes

EMQX Operator

EMQX クラスター管理

API リファレンス

パスワード認証

監視

アクセスコントロール

統合

管理

Prometheus との統合

OpenTelemetry との統合

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

プラグインカタログ

JT/T 808 ゲートウェイ

GB/T 32960 ゲートウェイ

Volcengine RTC

シナリオ

GPT-Realtime

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

動作の仕組み ​

特長と利点 ​

はじめる前に ​

前提条件 ​

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

データテーブルの作成 ​

コネクターの作成 ​

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

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

ルールのテスト ​

詳細設定 ​

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

動作の仕組み

特長と利点

はじめる前に

前提条件

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

データテーブルの作成

コネクターの作成

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

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

ルールのテスト

詳細設定