Apache IoTDBへのMQTTデータ取り込み

Apache IoTDBは、多種多様なIoTデバイスやシステムから生成される大量の時系列データを効率的に処理するために設計された、高性能かつスケーラブルな時系列データベースです。

EMQXはApache IoTDBとのシームレスなデータ統合を提供しており、EMQXでリアルタイムに受信したMQTTメッセージをREST API V2を通じてIoTDBに転送できます。この統合は一方向のデータフローをサポートし、MQTTデータをIoTDBに書き込むことで効率的な時系列データの保存と分析を可能にします。

本ページでは、EMQXとApache IoTDBの統合方法を紹介し、統合の作成および検証手順をステップバイステップで説明します。

動作概要

Apache IoTDBデータ統合は、追加のコーディングなしでMQTTベースの時系列データをApache IoTDBに取り込むことを可能にするEMQXの組み込み機能です。EMQXの組み込みルールエンジンを活用することで、データのフィルタリング、変換、転送を簡素化し、IoTDBでの効率的な保存とクエリを実現します。

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

データ統合のワークフローは以下の通りです：

メッセージのパブリッシュと受信：デバイスはMQTT経由でEMQXに接続し、テレメトリデータ、ステータス更新、イベント情報を含むメッセージをパブリッシュします。ルールエンジンが受信メッセージを評価します。
ルールベースの処理：定義されたルールにマッチしたメッセージが選択され、必要に応じてフィールドのフィルタリング、データ形式の変換、ペイロードの強化などの変換が適用されます。
データバッファリング：信頼性向上のため、IoTDBが一時的に利用不可の場合、EMQXはメッセージをメモリにバッファします。必要に応じてメモリ圧迫を避けるためにディスクにオフロード可能です。統合やEMQXノードの再起動時にはバッファデータは保持されません。
IoTDBへのデータ取り込み：マッチしたルールに対して、EMQXはIoTDB Sinkをトリガーし、処理済みデータを転送して時系列データとしてIoTDBに書き込みます。
データの保存と活用：IoTDBに保存されたデータは、デバイス監視、資産追跡、予知保全、運用最適化などの下流アプリケーションでクエリや分析に利用できます。

特長と利点

IoTDBとのデータ統合は、効果的なデータ処理と保存を実現するために以下の特長と利点を提供します：

ノーコードIoTデータパイプライン
EMQXの組み込みルールとSinkを使い、カスタムコードや外部サービスなしでMQTTから時系列データへの完全なパイプラインを構築可能です。
MQTTからIoTDBモデルへの柔軟なマッピング
TreeモデルとTableモデルの両方をサポートし、デバイスモデリングやクエリ要件に合わせた構造でMQTTデータをIoTDBに書き込めます。
取り込みと保存の分離
EMQXはバースト的で高頻度なMQTTトラフィックを吸収し、IoTDBは耐久性のある時系列保存に専念することで、システムの安定性とレジリエンスを向上させます。
本番対応のスケーラビリティ
デバイス数やデータ量に応じて水平スケール可能で、大規模なIoT、IIoT、エネルギー分野に適しています。
分析対応の時系列データ
IoTDBに書き込まれたデータは直接クエリ、集計、分析可能であり、ビッグデータエンジンと連携した高度な分析や長期的なインサイト取得にも活用できます。

はじめる前に

このセクションでは、EMQXダッシュボードでApache IoTDBデータ統合を作成する前に必要な準備について説明します。

前提条件

EMQXデータ統合のルールの知識
データ統合の知識

Apache IoTDBサーバーの起動

ここではDockerを使ったApache IoTDBサーバーの起動方法を紹介します。IoTDBの設定でenable_rest_service=trueが有効になっていることを確認してください。

以下のコマンドを実行して、RESTインターフェースを有効にしたApache IoTDBサーバーを起動します：

bash

docker run -d --name iotdb-service \
              --hostname iotdb-service \
              -p 6667:6667 \
              -p 18080:18080 \
              -e enable_rest_service=true \
              -e cn_internal_address=iotdb-service \
              -e cn_target_config_node_list=iotdb-service:10710 \
              -e cn_internal_port=10710 \
              -e cn_consensus_port=10720 \
              -e dn_rpc_address=iotdb-service \
              -e dn_internal_address=iotdb-service \
              -e dn_target_config_node_list=iotdb-service:10710 \
              -e dn_mpp_data_exchange_port=10740 \
              -e dn_schema_region_consensus_port=10750 \
              -e dn_data_region_consensus_port=10760 \
              -e dn_rpc_port=6667 \
              apache/iotdb:2.0.5-standalone

詳細はDocker HubのIoTDB実行方法をご参照ください。

データベースの作成

IoTDBはTreeモデルとTableモデルの2つのデータモデルをサポートしています。データベース作成前に、ConnectorおよびSinkで使用するSQL Dialect（TreeまたはTable）を確認し、それに応じてデータベースを作成してください。

Treeモデルの場合はデータベースのみ作成すればよいです。
Tableモデルの場合は、データベース作成後にテーブルを作成する必要があります。

詳細な手順はIoTDBユーザーガイドをご参照ください：

IoTDBコネクターの作成

Apache IoTDBデータ統合を作成するには、Apache IoTDB SinkとApache IoTDBサーバーを接続するためのコネクターを作成する必要があります。

EMQXはREST APIまたはThriftプロトコルを介してIoTDBと通信をサポートしています。

EMQXダッシュボードで Integrations -> Connectors に移動します。
右上の Create をクリックします。
Create Connector ページで Apache IoTDB を選択します。
コネクターを設定します：
- Connector Name：コネクターの一意な名前を入力します。大文字・小文字の英数字の組み合わせを使用してください。例：my_iotdb
- Description：（任意）コネクターの簡単な説明
- Driver：IoTDB接続に使用するプロトコルを選択します。
  - REST API：IoTDB RESTサービスのエンドポイント（例：http://localhost:18080）をIoTDB REST Service Base URLに入力します。
  - Thrift Protocol：IoTDB ThriftサーバーのアドレスをServer Hostに入力します。
- SQL Dialect：EMQXがIoTDBにデバイスデータを書き込む際のデータモデルを選択します。
  - Tree Model：階層的な時系列パスとしてデータを書き込み、パスベースのデバイス・計測管理に適しています。
  - Table Model：リレーショナルテーブルにデータを書き込み、デバイスタイプやカテゴリ別の管理に適しています。
- Database Name：SQL DialectがTable Modelの場合、接続するデータベース名を指定します。
- Username と Password：EMQXがApache IoTDBサーバーに認証するための資格情報を入力します。
- IoTDB Version：Apache IoTDBのバージョンを選択します。
- Enable TLS：Apache IoTDBサーバーへの暗号化接続を有効にします。詳細は外部リソースアクセスのTLSを参照してください。
- 任意のチューニングは高度な設定のAdvanced Settingsをご覧ください。
（任意）Test Connectivityをクリックして、コネクターがApache IoTDBサーバーに正常に接続できるか確認します。
Createをクリックしてコネクターの作成を完了します。
表示されるダイアログで、Back to Connector List または Create Rule を選択して、ルールとApache IoTDB Sinkの設定を続けられます。詳細はルールとApache IoTDB Sinkの作成をご覧ください。

Apache IoTDB Sinkを用いたルールの作成

このセクションでは、EMQXでMQTTのソーストピックroot/#からメッセージを処理し、処理結果を設定済みのApache IoTDB Sink経由でApache IoTDBに時系列データとして保存するルールの作成方法を説明します。

SQLを定義したルールの作成

EMQXダッシュボードで Integration -> Rules に移動します。
ページ右上の Create をクリックします。
ルールIDを入力します。例：my_rule
SQL editorに以下のステートメントを入力します。これはトピックパターンroot/#にマッチするMQTTメッセージを転送します：
sql
```
SELECT
  *
FROM
  "root/#"
```
TIP
初心者の方は、SQL ExamplesやEnable TestをクリックしてSQLルールの学習やテストが可能です。
処理結果をIoTDBに書き込むためにApache IoTDB Sinkをルールに追加します。詳細はApache IoTDB Sinkの追加をご覧ください。
Create Ruleページで設定内容を確認し、Saveをクリックしてルールを作成します。

ルール作成後、Rules一覧に表示されます。**Actions (Sink)**タブをクリックすると、このルールに紐づくIoTDB Sinkを確認できます。

また、Integrations -> Flow Designerに移動すると、トポロジーグラフが表示され、トピックroot/#のメッセージがmy_ruleルールで処理されIoTDBに書き込まれている様子を確認できます。

Apache IoTDB Sinkの追加

ルールの右側にあるAdd Actionボタンをクリックし、ルールにマッチした際にトリガーされるアクションを定義します。このアクションは処理済みデータをIoTDBに転送します。
Type of ActionドロップダウンからApache IoTDBを選択し、ActionはデフォルトのCreate Actionのままにします。既存のIoTDB Sinkを選択することも可能ですが、この例では新規作成を想定しています。
Sinkの名前と説明を入力します。
Connectorドロップダウンから先ほど作成したコネクターmy_iotdbを選択します。利用可能なコネクターがない場合は隣のボタンから作成できます。詳細はIoTDBコネクターの作成を参照してください。
Sinkの設定を行います：
- SQL Dialect：Apache IoTDB SinkがIoTDBにデータを書き込む方法を選択します。Connectorで選択したSQL Dialectと一致させる必要があります。
  - Tree Model：IoTDBの時系列パスとしてデータを書き込みます。各Sinkレコードはデバイスパスに挿入され、計測はそのデバイス下の個別時系列として書き込まれます。このモデル選択時はDevice IDフィールドを指定可能です。
  - Table Model：IoTDBのリレーショナルテーブルにデータを書き込みます。各Sinkレコードは指定テーブルの行として挿入され、フィールドはテーブルの列にマッピングされます。このモデル選択時はTableフィールドの指定が必須です。
- Device ID（任意）：IoTDBインスタンスに時系列データを転送・挿入する際のデバイス名として使用する特定のデバイスIDを入力します。
  TIP
  空欄の場合でも、パブリッシュされたメッセージ内やルール内でデバイスIDを指定可能です。例えば、JSONエンコードされたメッセージにdevice_idフィールドがあれば、その値が出力デバイスIDになります。ルールエンジンで抽出する場合は以下のようなSQLを使えます：
  sql
```
SELECT
 payload,
 `my_device` as payload.device_id
```
  ただし、このフィールドに設定した固定のデバイスIDが優先されます。
- Table：データを書き込むIoTDBのテーブル名を指定します。
- Align Timeseries：デフォルトは無効です。有効にすると、グループ化されたアラインド時系列のタイムスタンプ列がIoTDBに一度だけ保存され、個々の時系列で重複保存されません。詳細はAligned timeseriesを参照してください。
- Write Dataの設定で、MQTTメッセージからIoTDBデータを生成する方法を指定します。
  Write Dataセクションでは必要な数だけ項目を含むテンプレートを定義でき、各行に必要なコンテキスト情報を指定します。このテンプレートを用いてMQTTメッセージからIoTDBデータが生成されます。書き込みテンプレートはCSVファイルによる一括設定もサポートしています。詳細はバッチ設定をご覧ください。
  例として以下のテンプレートを考えます：
  注意
  Column CategoryはSQL DialectでTable Modelを選択した場合のみ表示されます。
  Column Category Timestamp Measurement Data Type Value
  field index INT32 ${index}
  temperature FLOAT ${temp}
  TimestampとValueはプレースホルダー構文をサポートし、変数で埋められます。Timestampが省略された場合は現在のシステム時刻（ミリ秒）が自動入力されます。
  その場合、MQTTメッセージは以下のような構造になります：
  json
```
{
  "index": "42",
  "temp": "32.67"
}
```
Fallback Actions：（任意）メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。これらはプライマリSinkがメッセージ処理に失敗した場合にトリガーされます。詳細はフォールバックアクションを参照してください。
Advanced settings：（任意）高度な設定をご覧ください。
（任意）Test ConnectivityをクリックしてSinkがApache IoTDBサーバーに接続できるかテストします。

Column Category	Timestamp	Measurement	Data Type	Value
field		index	INT32	${index}
		temperature	FLOAT	${temp}

バッチ設定

Apache IoTDBでは、ダッシュボード上で数百件のデータを同時に書き込む設定は困難な場合があります。これに対応するため、EMQXはデータ書き込みのバッチ設定機能を提供しています。

Write Dataの設定時に、CSVファイルから挿入操作用のフィールドを一括インポートできます。

Write DataテーブルのBatch Settingボタンをクリックし、Import Batch Settingポップアップを開きます。

指示に従いバッチ設定テンプレートファイルをダウンロードし、テンプレートに書き込み設定を記入します。デフォルトのテンプレート内容は以下の通りです：

注意

以下はTable Model用のデフォルトテンプレートです。Tree ModelではColumn Category列はありません。

Column Category	Timestamp	Measurement	Data Type	Value	備考（任意）
tag	now	clientid	text	${clientid}
field	now	temp	float	${payload.temp}	フィールド、値、データ型は必須。データ型はboolean, int32, int64, float, double, textが利用可能
attribute	now	hum	text	${payload.hum}
attribute	now	status	text	${payload.status}

Column Category：列のデータモデル。tag、field、attributeがサポートされます。tagは文字列である必要があり、fieldまたはattributeが推奨されます。
Timestamp：${var}形式のプレースホルダーをサポートし、タイムスタンプ形式が必要です。以下の特殊文字でシステム時刻を挿入可能です：
- now：現在のミリ秒タイムスタンプ
- now_ms：現在のミリ秒タイムスタンプ
- now_us：現在のマイクロ秒タイムスタンプ
- now_ns：現在のナノ秒タイムスタンプ
Measurement：フィールド名
Data Type：データ型。boolean, int32, int64, float, double, textが選択可能
Value：書き込むデータ値。定数または${var}形式のプレースホルダーをサポートし、データ型と一致する必要があります。
備考：CSVファイル内のメモ用で、EMQXへのインポートには使用されません。

1MB以下かつ2000行以内のCSVファイルのみサポートされます。

記入済みテンプレートファイルを保存し、Import Batch SettingポップアップにアップロードしてImportをクリックし、バッチ設定を完了します。
インポート後、Write Dataテーブルでさらにデータを調整可能です。

ルールのテスト

EMQXダッシュボード内蔵のWebSocketクライアントを使って、Apache IoTDB Sinkとルールの動作をテストできます。

ダッシュボード左のメニューで Diagnose -> WebSocket Client をクリックします。
現在のEMQXインスタンスの接続情報を入力します。
- ローカルでEMQXを実行している場合はデフォルト値を使用可能です。
- 認証設定を変更している場合は、ユーザー名やパスワードを入力してください。
ConnectをクリックしてクライアントをEMQXに接続します。
下にスクロールしてパブリッシュエリアに移動し、メッセージ内にデバイスIDを指定して以下を入力します：
- Topic：root/sg27
  TIP
  トピックがrootで始まらない場合、自動的にroot.がプレフィックスされます。例えばtest/sg27にメッセージをパブリッシュすると、デバイス名はroot.test.sg27になります。ルールとトピックの設定を正しく行い、そのトピックのメッセージがSinkに転送されるようにしてください。
- Payload：
  json
```
{
  "value": "37.6",
  "device_id": "root.sg27"
}
```
  TIP
  Write Dataテンプレートは以下の通りです：
```
now, "temp", float, "${payload.value}"
```
- QoS：2
Publishをクリックしてメッセージを送信します。
Sinkとルールが正常に作成されていれば、メッセージは指定したApache IoTDBの時系列テーブルにパブリッシュされているはずです。
IoTDBのコマンドラインインターフェースを使ってメッセージを確認します。上記のDocker環境を使っている場合は、以下のコマンドでサーバーに接続できます：
shell
```
$ docker exec -ti iotdb-service /iotdb/sbin/start-cli.sh -h iotdb-service
```

コンソールで以下を入力します：

sql

IoTDB> select * from root.sg27

以下のようにデータが表示されるはずです：

+------------------------+--------------+
|                    Time|root.sg27.temp|
+------------------------+--------------+
|2023-05-05T14:26:44.743Z|          37.6|
+------------------------+--------------+

高度な設定

このセクションでは、コネクターのパフォーマンスを最適化し、特定のシナリオに応じた動作をカスタマイズするための高度な設定オプションを説明します。コネクター作成時にAdvanced Settingsを展開し、ビジネスニーズに応じて以下の設定を行えます。

項目	説明	推奨値
HTTP Pipelining	サーバーに対して応答を待たずに連続して送信できるHTTPリクエストの最大数を指定します。正の整数値で設定します。 `1`の場合は従来のリクエスト-レスポンスモデルとなり、各リクエスト送信後に応答を待ちます。値を大きくすると複数リクエストをバッチ送信でき、ネットワークリソースの効率的利用とラウンドトリップ時間の短縮が可能です。	`100`
Pool Type	EMQXとApache IoTDB間の接続管理・分配に使用するアルゴリズム戦略を定義します。 `random`の場合、利用可能な接続プールからランダムに接続を選択し、シンプルでバランスの良い分配を行います。 `hash`の場合、ハッシュアルゴリズムでリクエストを一貫してプール内の接続にマッピングします。クライアントIDやトピック名に基づくロードバランシングなど、決定的な分配が必要な場合に適しています。注意：適切なプールタイプはユースケースや求める分配特性によります。	`random`
Connection Pool Size	Apache IoTDBサービスとの接続プールで維持可能な同時接続数を指定します。システムのスケーラビリティとパフォーマンス管理に役立ちます。注意：適切なサイズはシステムリソース、ネットワークレイテンシ、ワークロードに依存します。大きすぎるとリソース枯渇、小さすぎるとスループット制限の原因となります。	`8`
Connect Timeout	EMQXがApache IoTDB HTTPサーバーへの接続確立を試みる際の最大待機時間（秒）を指定します。注意：適切なタイムアウト設定はシステムパフォーマンスとリソース利用のバランスに重要です。様々なネットワーク条件でテストし最適値を見つけてください。	`15`
HTTP Request Max Retries	EMQXとApache IoTDB間の通信でHTTPリクエストが失敗した場合に再試行する最大回数を指定します。	`2`
Start Timeout	自動起動されたリソースが正常状態になるまで待機する最大時間（秒）を指定します。リソース作成リクエストに対する応答前に、接続先リソース（例：Apache IoTDBのデータベースインスタンス）が完全に稼働していることを確認するための設定です。	`5`
Buffer Pool Size	EMQXとApache IoTDB間のイーグレス型ブリッジでデータフロー管理に割り当てるバッファワーカープロセス数を指定します。これらのワーカーは送信前のデータを一時的に保持・処理します。イーグレス（送信）専用のブリッジに関連し、イングレス（受信）専用の場合は`0`に設定可能です。	`18`
Request TTL	バッファに入ったリクエストが有効とみなされる最大期間（秒）を指定します。バッファリング開始時からカウントし、TTLを超えたリクエストや送信後にタイムリーな応答・アックが得られないリクエストは期限切れとみなされます。	`45`
Health Check Interval	Apache IoTDBとの接続に対して自動ヘルスチェックを行う間隔（秒）を指定します。	`15`
Max Buffer Queue Size	Apache IoTDBデータ統合における各バッファワーカーがバッファ可能な最大バイト数を指定します。バッファワーカーはデータ送信前に一時保持し、データフローの効率化に寄与します。システム性能やデータ転送要件に応じて調整してください。	`265`
Query Mode	メッセージ送信の要件に応じて`asynchronous`（非同期）または`synchronous`（同期）クエリモードを選択します。非同期モードではIoTDBへの書き込みがMQTTメッセージのパブリッシュ処理をブロックしませんが、クライアントがIoTDBへの到達前にメッセージを受信する可能性があります。	`Async`
Inflight Window	「インフライトクエリ」とは開始済みで応答・アック未受領のクエリを指します。ConnectorがApache IoTDBと通信する際に同時に存在可能なインフライトクエリの最大数を制御します。 `query_mode`が`async`の場合、このパラメータは特に重要です。同一MQTTクライアントからのメッセージを厳密に順序処理したい場合は`1`に設定してください。	`100`

さらに詳しく

EMQXはApache IoTDBとのデータ統合に関する豊富な学習リソースを提供しています。以下のリンクから詳細を学べます：

ブログ：

IoT向け時系列データベース（TSDB）：欠けていたピース

Kubernetes

EMQX Operator

EMQX クラスター管理

API リファレンス

パスワード認証

監視

アクセスコントロール

統合

管理

Prometheus との統合

OpenTelemetry との統合

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

プラグインカタログ

JT/T 808 ゲートウェイ

GB/T 32960 ゲートウェイ

Volcengine RTC

シナリオ

GPT-Realtime

Apache IoTDBへのMQTTデータ取り込み ​

動作概要 ​

特長と利点 ​

はじめる前に ​

前提条件 ​

Apache IoTDBサーバーの起動 ​

データベースの作成 ​

IoTDBコネクターの作成 ​

Apache IoTDB Sinkを用いたルールの作成 ​

SQLを定義したルールの作成 ​

Apache IoTDB Sinkの追加 ​

バッチ設定 ​

ルールのテスト ​

高度な設定 ​

さらに詳しく ​