EMQX TablesへのMQTTデータ取り込み

EMQX Tablesは、EMQX Cloudに組み込まれたネイティブでフルマネージドの時系列データストレージサービスです。高スループットかつ低レイテンシでのMQTTデータの取り込みと分析に最適化されており、IoTユースケースに理想的です。

GreptimeDBを基盤としており、EMQX ブローカーとシームレスに統合され、InfluxDB Line Protocolをサポートすることで、テレメトリデータの効率的な保存、クエリ、可視化を実現します。

詳細はEMQX Tables概要をご覧ください。

本ガイドでは、以下の手順でMQTTデータをEMQX Tablesに取り込む方法を説明します。

• コネクターの作成
• EMQX Tablesアクションを持つルールの作成
• データ取り込みのテストと結果のクエリ

前提条件

以下のデプロイメントを作成済みであること：

• EMQX ブローカーのデプロイメント：サーバレス、[専用]または専用フレックス
• EMQX Tablesのデプロイメント

また、以下に精通していること：

• EMQXルール
• EMQXのデータ統合
• InfluxDB Line Protocol

ブローカーとTables間の接続

接続方法は、ブローカーのデプロイメントタイプとネットワーク構成によって異なります。

デプロイメントタイプ	条件	接続方法	必要な対応
専用 / 専用フレックス	Tablesと同じクラウドプラットフォーム、リージョン、ネットワーク	プライベート（セキュアかつ低レイテンシ）	プロジェクトレベルネットワーク管理でネットワーク共有を設定
専用 / 専用フレックス	Tablesと異なるリージョンまたはネットワーク	TLS経由のパブリックインターネット	ブローカーのデプロイメントのネットワーク管理設定でNATゲートウェイを有効化
サーバレス	—	TLS経由のパブリックインターネット	対応不要。ネットワーク関連の設定（ネットワーク関連付けやNATゲートウェイ）は適用されません

EMQX Tablesコネクターの作成

データを書き込む前に、EMQX Tablesへのコネクターを作成します。

新しいEMQX Tablesデプロイメントでは、デプロイメント詳細にデフォルトのユーザー名やパスワードが返されないため、コネクター設定前にEMQX Tablesユーザーを作成してください。ユーザー作成時はBroker Integrationプリセットを選択します。これによりコネクター設定に必要な権限が付与されます。コネクター設定中に権限検証が失敗した場合は、エラーメッセージに従ってEMQX Tablesユーザーの権限を更新してください。詳細はユーザー管理をご参照ください。

1. EMQX ブローカーのデプロイメントにアクセスし、左メニューからデータ統合をクリックします。

2. 初めてコネクターを作成する場合はEMQX Tablesを探します。既にコネクターを作成済みの場合は、+ 新規コネクターをクリックし、EMQX Tablesを選択します。

3. 新規コネクターページで、コネクター名は自動生成されます。以下の2つのセットアップモードから選択してください。

   クイックセットアップでは接続情報が自動入力されます。同一プロジェクト内のすべてのEMQX Tablesデプロイメントがドロップダウンに表示されます。

   1. クイックセットアップを選択（デフォルト）。

   2. EMQX Tablesデプロイメントの選択：ドロップダウンからTablesデプロイメントを選択します。選択したデプロイメントがブローカーと異なるネットワークまたはリージョンにある場合、接続を有効にするためにブローカーのNATゲートウェイを有効化するよう通知が表示されます。

   3. データベース名：デフォルトのpublicデータベースを使用するか、カスタムデータベース名を指定します。指定したデータベースはEMQX Tablesデプロイメント内に既に作成されている必要があります。詳細はカスタムデータベースの作成をご覧ください。

   4. ユーザー名：選択したEMQX Tablesデプロイメントのユーザーを選択します。ドロップダウンにはデプロイメント内のすべてのユーザー名が表示され、権限やデータベースアクセスによるフィルターはかかりません。

      適切なユーザーが存在しない場合は、ユーザー名ドロップダウンのユーザー追加をクリックしてEMQX Tablesのユーザー管理ページを開き、作成してください。Broker Integrationプリセットで作成したユーザーを使用可能です。詳細はユーザー管理をご参照ください。

      

      選択したユーザーは対象データベースへのアクセス権を持ち、EMQX Tablesのユーザー権限要件を満たしている必要があります。データベース名を入力または変更すると、EMQX Cloudが選択ユーザーに対して検証を行います。ユーザーにアクセス権がない場合、データベース名欄にエラーが表示されます。

      

   5. パスワード：選択したユーザーのパスワードを入力します。

   1. EMQX Tablesデプロイメントが異なるプロジェクトにある場合や、手動で接続設定を行いたい場合はカスタムセットアップを選択します。

      専用および専用フレックスのデプロイメントでは、ブローカーのネットワーク管理でパブリックアクセス用のNATゲートウェイを有効にする必要があります。

   2. 接続情報を入力します：

      項目	説明	例
      パブリックホスト	ホストアドレスとポート。パブリックインターネット接続の場合、ポート5001（gRPCS）を使用し、TLSを有効にするをオンにすることを推奨します。	your-emqx-tables-public-host:5001
      データベース名	対象データベース名。既に存在し、指定ユーザーがアクセス権を持っている必要があります。詳細はカスタムデータベースの作成をご参照ください。	public またはカスタムデータベース名
      ユーザー名	対象データベースにアクセス権を持ち、EMQX Tablesのユーザー権限要件を満たすユーザー名	supervisor
      パスワード	指定ユーザーのパスワード	******

      

4. テストをクリックして接続を検証します。EMQX Tablesサービスにアクセス可能で、ユーザー名とパスワードが有効かつ選択ユーザーが必要な権限とデータベースアクセスを持っていれば、成功メッセージが表示されます。

5. 新規作成をクリックして作成を完了します。これで、このコネクターを用いたルール作成に進めます。

EMQX Tablesへのデータ取り込み用ルールの作成

次に、書き込み対象のデータを指定し、EMQX Tablesへ書き込むアクションを追加するルールを作成します。

1. ルールセクションで新規ルールをクリックするか、コネクター横のアクションアイコンを使用します。

2. SQLエディターでSQLルールを定義します。本例では、クライアントがtemp_hum/emqxトピックに温度と湿度のメッセージを送信した際にエンジンをトリガーすることを目的としています。以下のようにSQLを設定します。

     SELECT
       timestamp,
       payload.location as location,
       payload.temp as temp,
       payload.hum as hum
     FROM "temp_hum/emqx"

   TIP

   初心者の方はTry It OutをクリックしてSQLルールの学習とテストを行うことができます。

3. 次へをクリックしてルールにアクションを追加します。

4. 新規アクション（シンク）ページで、先ほど作成したコネクターがコネクター欄にデフォルト選択されています。

5. 時間精度をmillisecond（デフォルト）に設定します。

6. データフォーマットを設定し、EMQX Tablesへのデータ解析・書き込み方法を定義します。デフォルトのフォーム形式を使うか、書き込み構文を手動で入力したい場合はLine Protocolを選択します。

   • フォーム：構造化された設定フィールドを好む場合に使用します。EMQXが設定されたmeasurement、timestamp、タグ、フィールドをline protocol形式に変換してEMQX Tablesに書き込みます。

     以下のように設定します：

     項目	説明	例
     Measurement	書き込み先のmeasurement名。プレースホルダー対応。	temp_hum
     Timestamp	データポイントのタイムスタンプ（任意）。空欄または${timestamp}の場合、EMQXはホストマシンのUTCシステム時刻を使用します。メッセージ内のカスタムタイムスタンプを使う場合は、選択した時間精度と一致していることを確認してください。	${timestamp}
     Tags	インデックス付きカラムの任意のキー・バリュー。タグカラムは高速フィルタリングのためにデフォルトでインデックス付きで、SQLではPRIMARY KEYカラムとして宣言されます。	キー: location 値: ${location}
     Fields	書き込み対象の必須キー・バリュー。キー・値ともにプレースホルダー対応。数値はデフォルトでfloatとして書き込まれます。別の型を指定する場合は型サフィックスを付けます（例：${payload.int_key}i）。インポートをクリックしてフィールドマッピングを一括インポートも可能です。	キー: temp 値: ${temp} キー: hum 値: ${hum}

   • Line Protocol：最終的な書き込み構文を完全に制御したい場合に選択します。書き込み構文に、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）。詳細は同上リンクを参照してください。

7. 確認をクリックしてルールを保存します。

8. 新規ルール作成成功ポップアップでルールに戻るをクリックし、ルール作成を完了します。

ルールのテストとデータのクエリ

MQTTXなどのクライアントツールを使って温度・湿度データの送信をシミュレートすることを推奨します。簡単なデモには、ブローカーのデプロイメント内にある組み込みの診断ツールを使い、左メニューのオンラインテストをクリックしてください。

1. オンラインテストで、ユーザー名とパスワード、または自動生成認証を使ってデプロイメントに接続します。

2. メッセージセクションで以下のメッセージを送信します：

   • トピック：temp_hum/emqx

   • ペイロード：

     {
       "temp": 27.5,
       "hum": 41.8,
       "location": "Prague"
     }

   

3. EMQX Tablesのデプロイメントに移動し、左メニューからデータエクスプローラーをクリックします。

4. 以下のSQLを実行し、publicテーブルに取り込まれたデータをクエリします。

   select * from "temp_hum"

   クエリ結果テーブルに1件のレコードが表示されるはずです。

   

5. EMQX ブローカーのデプロイメントでルール統計を確認します。ルール一覧でルールIDをクリックすると、そのルールおよび関連アクションの統計情報が表示されます。