EMQX TablesへのMQTTデータ取り込み
EMQX Tablesは、EMQX Cloudに組み込まれたネイティブで完全管理された時系列データストレージサービスです。高スループットかつ低レイテンシでのMQTTデータの取り込みと分析に最適化されており、IoT（モノのインターネット）ユースケースに理想的です。
GreptimeDBを基盤とし、EMQX TablesはEMQXブローカーとシームレスに統合され、InfluxDB Line Protocolをサポートすることで、テレメトリデータの効率的な保存、クエリ、可視化を可能にします。
詳細はEMQX Tables概要をご覧ください。
コネクター、ルール、アクションの作成と管理を案内するワークフローにはMessage Historyを利用してください。Message HistoryはEMQX v6.1.4以降を実行するDedicatedおよびDedicated Flexデプロイメントでのみ利用可能です。Serverlessデプロイメント、その他のネットワーク構成、カスタムRule SQLや複数アクション・宛先、統合リソースの独立制御が必要な場合は、本Data Integrationワークフローを継続してご利用ください。
本ガイドでは、MQTTデータをEMQX Tablesに取り込む手順として以下を説明します。
- コネクターの作成
- EMQX Tablesアクションを持つルールの作成
- データ取り込みとクエリ結果のテスト
前提条件
以下のデプロイメントを作成済みであること：
- EMQXブローカーのデプロイメント：Serverless、DedicatedまたはDedicated Flex
- EMQX Tablesのデプロイメント
以下に習熟していること：
ブローカーとTables間の接続
接続方法はブローカーのデプロイメントタイプとネットワーク構成によって異なります：
| デプロイメントタイプ | 条件 | 接続方法 | 必要な操作 |
|---|---|---|---|
| Dedicated / Dedicated Flex | Tablesと同じクラウドプラットフォーム、リージョン、ネットワーク | プライベート（セキュアかつ低レイテンシ） | プロジェクトレベルネットワーク管理でネットワーク共有を設定 |
| Dedicated / Dedicated Flex | Tablesと異なるリージョンまたはネットワーク | TLS経由のパブリックインターネット | ブローカーのネットワーク管理設定でNATゲートウェイを有効化 |
| Serverless | — | TLS経由のパブリックインターネット | 操作不要。ネットワーク関連設定やNATゲートウェイは不要 |
EMQX Tablesコネクターの作成
データ書き込みの前に、EMQX Tablesへのコネクターを作成します。
新しいEMQX Tablesデプロイメントでは、デプロイメント詳細にデフォルトのユーザー名やパスワードが返されないため、コネクター設定前にEMQX Tablesユーザーを作成してください。ユーザー作成時にBroker Integrationプリセットを選択すると、コネクター設定に必要な権限が付与されます。コネクター設定時に権限検証が失敗した場合は、エラーメッセージに従いEMQX Tablesユーザーの権限を更新してください。詳細はユーザー管理を参照してください。
EMQXブローカーのデプロイメントに移動し、左メニューからデータ統合をクリックします。
初めてコネクターを作成する場合はEMQX Tablesを探します。既にコネクターを作成済みの場合は**+ 新規コネクター**をクリックし、EMQX Tablesを選択します。
新規コネクターページで、コネクター名は自動生成されます。セットアップモードを2種類から選択してください。
テストをクリックして接続を検証します。EMQX Tablesサービスにアクセス可能で、ユーザー名とパスワードが有効、かつ選択ユーザーが必要な権限とデータベースアクセスを持っていれば成功メッセージが表示されます。
新規作成をクリックして作成を完了します。これでこのコネクターを使ったルール作成に進めます。
EMQX Tablesへのデータ取り込み用ルールの作成
次に、書き込むデータを指定し、EMQX Tablesへの書き込みアクションを追加するルールを作成します。
ルールセクションで新規ルールをクリックするか、コネクター横のアクションアイコンを使用します。
SQLエディターでSQLルールを定義します。ここでは、クライアントが
temp_hum/emqxトピックに温度と湿度のメッセージを送信した際にエンジンをトリガーする例を示します。SQLは以下のように設定します。sqlSELECT timestamp, payload.location as location, payload.temp as temp, payload.hum as hum FROM "temp_hum/emqx"TIP
初心者の方はTry It OutをクリックしてSQLルールの学習とテストを行ってください。
次へをクリックしてルールにアクションを追加します。
新規アクション（シンク）ページで、先ほど作成したコネクターがコネクター欄にデフォルト選択されています。
Time Precisionを
millisecond（デフォルト）に設定します。データフォーマットを設定し、データの解析およびEMQX Tablesへの書き込み方法を定義します。デフォルトのフォーム形式を使うか、書き込み構文を手動で入力したい場合はLine Protocolを選択できます。
フォーム：構造化された設定フィールドを好む場合に使用します。EMQXは設定されたmeasurement、timestamp、タグ、フィールドをLine Protocol形式に変換してEMQX Tablesに書き込みます。
各フィールドの設定例：
項目 説明 例 Measurement 書き込むmeasurement名。プレースホルダーをサポート。 temp_humTimestamp データポイントのタイムスタンプ（任意）。空欄または ${timestamp}の場合、EMQXはホストマシンのUTCシステム時刻を使用。メッセージ内のカスタムタイムスタンプを使う場合は、選択したTime Precisionに合致していることを確認。${timestamp}Tags インデックス付きカラムの任意のキー・バリュー。タグカラムは高速フィルタリングのためデフォルトでインデックス化され、SQLでは PRIMARY KEYカラムとして宣言される。キー： location
値：${location}Fields 書き込むデータフィールドの必須キー・バリュー。キー・値ともにプレースホルダーをサポート。数値はデフォルトでfloatとして書き込まれる。別の型を指定する場合は型サフィックスを付ける（例： ${payload.int_key}i）。インポートをクリックしてフィールドマッピングを一括インポートも可能。キー： temp
値：${temp}
キー：hum
値：${hum}Line Protocol：最終的な書き込み構文を完全に制御したい場合に選択します。Write Syntaxにテキストベースのフォーマットを指定し、measurement、タグセット、フィールドセット、タイムスタンプおよびサポートされるプレースホルダーを記述します。詳細はInfluxDB 2.3 Line ProtocolおよびInfluxDB 1.8 Line Protocolを参照してください。
例：
temp_hum,location=${location} temp=${temp},hum=${hum} ${timestamp}TIP
- 符号付き整数型の値を書き込む場合は、プレースホルダーの後に
iを付けます（例：${payload.int}i）。詳細はInfluxDB 1.8の整数値書き込みを参照。 - 符号なし整数型の値を書き込む場合は、プレースホルダーの後に
uを付けます（例：${payload.int}u）。詳細は同上リンクを参照。
- 符号付き整数型の値を書き込む場合は、プレースホルダーの後に
確認をクリックしてルールを保存します。
新規ルール成功ポップアップでルールに戻るをクリックし、ルール作成を完了します。
ルールのテストとデータのクエリ
温度・湿度データの送信をシミュレートするために、MQTTXなどのクライアントツールの利用を推奨します。簡単なデモには、ブローカーの左メニューからオンラインテストを使った組み込み診断ツールも利用可能です。
オンラインテストで、ユーザー名とパスワードまたは自動生成認証を使ってデプロイメントに接続します。
メッセージ欄で以下のメッセージを送信します：
トピック：
temp_hum/emqxペイロード：
json{ "temp": 27.5, "hum": 41.8, "location": "Prague" }
EMQX Tablesのデプロイメントに移動し、左メニューからデータエクスプローラーをクリックします。
以下のSQLを実行して、
publicテーブルに取り込まれたデータをクエリします：sqlselect * from "temp_hum"クエリ結果テーブルに1件のレコードが表示されるはずです。
EMQXブローカーのデプロイメントでルール統計を確認します。ルール一覧のルールIDをクリックすると、該当ルールおよび関連アクションの統計情報が表示されます。
