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

注意

Apache Doris とのデータ統合は、EMQX バージョン 5.10.0 以降の Dedicated または Dedicated Flex エディションで利用可能です。

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

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

注意

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

動作概要

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

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

IoT デバイスが EMQX Cloud に接続：IoT デバイスが MQTT プロトコルで正常に接続されると、オンラインイベントがトリガーされます。イベントにはデバイス ID、送信元 IP アドレスなどの情報が含まれます。
メッセージのパブリッシュと受信：デバイスはテレメトリや状態データを特定のトピックにパブリッシュします。EMQX Cloud はこれらのメッセージを受信すると、ルールエンジン内でマッチング処理を開始します。
ルールエンジンによるメッセージ処理：組み込みのルールエンジンは、特定のトピックに基づいてメッセージやイベントを処理します。ルールエンジンは対応するルールをマッチングし、データ形式の変換、特定情報のフィルタリング、コンテキスト情報の付加などの処理を行います。
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 ツールやクライアントアプリケーションとの連携も容易で、ダッシュボード、レポート、自動化ワークフローに活用できます。
ランタイムメトリクス：各アクションの実行時メトリクス（総メッセージ数、成功/失敗数、現在の処理レートなど）を閲覧可能です。

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

はじめる前に

このセクションでは、EMQX Cloud コンソールで Apache Doris データ統合を作成する前に必要な準備について説明します。Apache Doris サーバーのインストールやデータテーブルの作成が含まれます。

前提条件

ルールの理解
データ統合の理解

ネットワーク設定

データ統合を構成する前に、EMQX Cloudのデプロイメントを作成し、EMQX Cloudと対象サービス間のネットワーク接続を確立していることを確認してください。

Dedicated Flexデプロイメントの場合：
EMQX CloudのVPCと対象サービスのVPC間でVPCピアリング接続を作成します。ピアリング接続が確立されると、EMQX Cloudは対象サービスのプライベートIPアドレスを介してアクセス可能になります。
パブリックIP経由でのアクセスが必要な場合は、NATゲートウェイを構成してアウトバウンド接続を有効にしてください。
BYOC（Bring Your Own Cloud）デプロイメントの場合：
BYOCデプロイメントが稼働しているVPCと対象サービスをホストするVPC間でVPCピアリング接続を作成します。ピアリングが確立されると、対象サービスのプライベートIPアドレスを介してアクセス可能になります。
対象サービスにパブリックIP経由でアクセスする必要がある場合は、クラウドプロバイダーのコンソールを使用してBYOC VPCにNATゲートウェイを構成してください。

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);

コネクターの作成

このセクションでは、EMQX Cloud から Apache Doris サーバーへデータを送信するためのコネクターの作成方法を説明します。

デプロイメント画面に移動し、左側のナビゲーションメニューから データ統合 をクリックします。
初めてコネクターを作成する場合は、データ永続化 カテゴリの中から Doris を選択します。既にコネクターを作成済みの場合は、新しいコネクター を選択し、続いて データ永続化 カテゴリの中から Doris を選択します。
新しいコネクター ページで以下の情報を設定します。
- コネクター名：システムが自動的に名前を生成します。
- サーバーホスト：127.0.0.1:9030 または Apache Doris サーバーがリモートの場合は実際のホスト名を入力します。
- データベース名：mqtt を入力します。
- ユーザー名：root を入力します。
- パスワード：public を入力します。
ビジネス要件に応じて詳細設定（任意）を行います。
新規作成 をクリックする前に、テスト ボタンを押してコネクターが Apache Doris サーバーに接続できるか確認できます。
画面下部の 新規作成 ボタンをクリックしてコネクターの作成を完了します。

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

このセクションでは、ソース MQTT トピック t/# からのメッセージを処理し、処理結果を設定済みのアクションを介して Apache Doris の emqx_messages テーブルに保存するルールの作成方法を説明します。

ルールエリアで 新しいルール をクリックするか、作成したコネクターの アクション 列にある新規ルールアイコンをクリックします。
ルール ID に my_rule と入力し、SQL エディター に以下の文を設定します。これはトピック t/# 配下の MQTT メッセージを Apache Doris に保存することを意味します。
注意：独自の SQL 構文を指定する場合は、アクションで必要なすべてのフィールドを SELECT 部分に含めていることを確認してください。
sql
```
SELECT
  *
FROM
  "t/#"
```
TIP
初心者の方は、SQL Examples と Try It Out をクリックして SQL ルールの学習とテストを行えます。
次へをクリックしてアクションを追加します。
コネクター ドロップダウンから先ほど作成したコネクターを選択します。
利用する機能に応じて SQL テンプレート を設定します。
注意：これは事前処理された SQL なので、フィールドは引用符で囲まず、文末にセミコロンを付けないでください。
sql
```
INSERT INTO emqx_messages(clientid, topic, payload, created_at) VALUES(
  ${clientid},
  ${topic},
  ${payload},
  FROM_UNIXTIME(${timestamp}/1000)
)
```
SQL テンプレート内でプレースホルダ変数が未定義の場合、SQL テンプレート 上部の Undefined Vars as Null スイッチでルールエンジンの動作を切り替えられます。
- 無効（デフォルト）：ルールエンジンは未定義変数に文字列 undefined をデータベースに挿入します。
- 有効：未定義変数の場合、ルールエンジンは NULL をデータベースに挿入します。
  TIP
  可能な限りこのオプションは有効にしてください。無効にするのは後方互換性確保時のみ推奨されます。
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。詳細はフォールバックアクションを参照してください。
確認ボタンをクリックしてアクション設定を完了します。
新しいルールが成功しました ポップアップで ルールに戻る をクリックし、データ統合の設定チェーンを完了します。

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

このセクションでは、クライアントのオンライン/オフライン状態を記録し、イベントデータを設定済みアクションを介して Apache Doris の emqx_client_events テーブルに保存するルールの作成方法を説明します。

ルール作成手順はメッセージ保存用ルールの作成とほぼ同様ですが、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つのアクションの稼働状況を確認してください。新しい受信メッセージと送信メッセージがそれぞれ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)

プランと料金

はじめに

デプロイメントの作成

専用フレックス

BYOC

ネットワーク管理

VPC ピアリング接続

PrivateLink の設定

クライアント認証

拡張認証

クライアント認可

拡張認可

モニタリング

ゲートウェイ

クライアント＆アプリケーション接続

データ統合

ルール

Flowデザイナー

LLM ベースの MQTT データ処理

スマートデータハブ

クエリガイド

統合ガイド

シングルサインオン（SSO）

プラットフォーム API

ブローカーデプロイメント API

サーバレスデプロイメント API

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

動作概要

特長とメリット

はじめる前に

前提条件

ネットワーク設定

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

データテーブルの作成

コネクターの作成

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

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

ルールのテスト

専用フレックス

VPC ピアリング接続

PrivateLink の設定

拡張認証

拡張認可

ルール

LLM ベースの MQTT データ処理

サーバレス デプロイメント API

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

動作概要 ​

特長とメリット ​

はじめる前に ​

前提条件 ​

ネットワーク設定 ​

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

データテーブルの作成 ​

コネクターの作成 ​

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

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

ルールのテスト ​

サーバレスデプロイメント API

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

動作概要

特長とメリット

はじめる前に

前提条件

ネットワーク設定

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

データテーブルの作成

コネクターの作成

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

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

ルールのテスト