EMQX Enterprise

日本語

简体中文
English

日本語

简体中文
English

Appearance

お問い合わせ
お問い合わせ

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

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

GreptimeDBを基盤とするEMQX Tablesは、EMQXブローカーとシームレスに統合され、InfluxDB Line Protocolをサポートすることで、テレメトリデータの効率的な保存、クエリ、および可視化を可能にします。詳細はEMQX Tables概要をご覧ください。

EMQX Enterprise 6.1以降では、EMQX TablesコネクターおよびSinkが提供されており、オンプレミスのEMQX Enterprise環境からEMQX Cloud上のEMQX TablesへMQTTデータを安全に書き込み、集中クエリおよび処理を行うことが可能です。

enterprise_tables_integration

本ページでは、EMQX EnterpriseからEMQX CloudのEMQX TablesへMQTTデータを取り込む手順を以下の流れで説明します。

  • EMQX EnterpriseとEMQX Tables間のネットワーク接続の確立
  • EMQX Tablesコネクターの作成
  • EMQX Tablesアクションを含むルールの作成
  • データ取り込みとクエリ結果のテスト

前提条件 ​

開始前に以下の条件を満たしていることを確認してください。

  • EMQX Enterprise バージョン6.1以降がオンプレミスまたはプライベート環境にデプロイされていること。

  • EMQX Cloudコンソール上にEMQX Tablesのデプロイメントが作成され、稼働していること。

    emqx_tables_connection_info

  • EMQX EnterpriseのデプロイメントからEMQX Tablesのエンドポイントへネットワーク経由で到達可能であること（パブリックエンドポイントまたはプライベート接続環境に応じて）。

  • 以下の内容に習熟していること：

EMQX Tablesコネクターの作成 ​

データ書き込みの前に、EMQX Enterprise環境でEMQX Tablesへのコネクターを作成します。

  1. EMQX Enterpriseダッシュボードで、データ統合 -> コネクターに移動します。

  2. + 新規コネクターをクリックし、EMQX Tablesを選択します。

  3. コネクター作成ページで以下の設定を行います。

    • コネクター名：コネクターの一意な名前を入力します。

    • 説明（任意）：識別用の簡単な説明を追加します。

    • サーバーホスト：<host>:<port>形式でEMQX Tablesサービスのアドレスを入力します。例：tables.example.emqx.com:4001

    • データベース：EMQX Tables内の対象データベース名を指定します。例：public

      TIP

      EMQX Tablesデプロイメント作成時に、デフォルトでpublicデータベースが作成されます。カスタムデータベースを作成したい場合はカスタムデータベースの作成を参照してください。

    • ユーザー名：EMQX Tablesデプロイメントで提供されたユーザー名を入力します。

    • パスワード：対応するパスワードを入力します。

    • TLSを有効化：EMQX Tablesへの接続時にTLS暗号化を使用する場合はこのオプションを有効にします。TLSは本番環境での利用を推奨します。

    • 詳細設定（任意）：接続プールサイズ、タイムアウト、リトライ動作などの高度なオプションを必要に応じて設定します。

  4. 接続テストをクリックし、接続可能か確認します。EMQX Tablesサービスに接続できれば成功メッセージが表示されます。

  5. 作成をクリックしてコネクター作成を完了します。

このコネクターはルールやアクション定義時に利用可能です。

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

次に、どのMQTTメッセージをどのようにEMQX Tablesに書き込むかを指定するルールを作成します。

SQLルールの定義 ​

  1. データ統合 -> ルールに移動します。

  2. + 作成をクリックします。

  3. SQLエディターでルールロジックを定義します。以下の例では、クライアントがtemp_hum/emqxトピックに温度と湿度データをパブリッシュした際にトリガーされます。

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

    TIP

    EMQXルールが初めての場合は、Try It OutをクリックしてSQLルールをインタラクティブに学習・テストできます。

  4. + アクション追加をクリックし、ルールにアクションを追加します。

EMQX Tablesアクションの追加 ​

SQLルールを定義した後、ルール発動時に選択されたデータをEMQX Tablesに書き込むアクションを追加します。

  1. アクションの種類でEMQX Tablesを選択します。

  2. アクションはアクション作成のままにします。

  3. 以下の項目を設定します。

    • 名前：アクションの名前を入力します。

    • コネクター：先ほど作成したEMQX Tablesコネクターを選択します。

    • 説明（任意）：このアクションの説明を追加します。

    • 書き込み構文：EMQX Tablesにデータを書き込むためのInfluxDB Line Protocol形式を定義します。

      書き込み構文内のプレースホルダー（例：${location}, ${temp}）はSQLルールで選択したフィールド名と対応している必要があります。ルールがトリガーされると、EMQXはこれらのプレースホルダーをSQLクエリで生成された値に置き換えます。

      行プロトコルの先頭のmeasurementがテーブル名として使用されます。データが初めて正常に書き込まれるとテーブルが自動作成されます。

      例：

      pgsql
      temp_hum,location=${location} temp=${temp},hum=${hum} ${timestamp}

      この例では：

      • temp_humがmeasurementでありテーブル名として使われます。
      • locationはタグとして書き込まれます。
      • tempとhumはフィールドとして書き込まれます。
      • ${timestamp}はルールエンジンによって生成されたタイムスタンプを提供します。

      注意：

      • 符号付き整数値を書き込む場合はプレースホルダーの後にiを付けます。例：${payload.int}i
      • 符号なし整数値の場合はuを付けます。例：${payload.int}u
      • サフィックスを付けない場合、整数値はデフォルトで符号付き整数として解釈され、小数点を含む値は浮動小数点数として扱われます。
      • 値が負の可能性があるか符号付き整数として保存する必要がある場合はiを使用し、非負の値で符号なし整数として保存したい場合（例：カウンター、ID、単調増加メトリクス）はuを使用してください。

    • 時間精度：タイムスタンプの時間精度を選択します。デフォルトはmillisecondです。

    • フォールバックアクション（任意）：このアクションが失敗した場合に実行するフォールバックアクションを設定できます。デフォルトでは設定されていません。詳細はフォールバックアクションを参照してください。

    • 詳細設定（任意）：バッチ処理やリトライポリシーなどの高度な動作を必要に応じて設定します。

    emqx_tables_action

  4. 作成をクリックしてアクションを保存します。

  5. ルール作成ページで保存をクリックしてルールを保存します。

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

MQTTXなどのクライアントツールを使って温度と湿度データの送信をシミュレーションすることを推奨します。簡単なデモとして、ダッシュボード内の組み込み診断ツールを利用することも可能です。

Websocketクライアントでテストデータをパブリッシュ ​

  1. EMQX Enterpriseダッシュボードの左メニューから診断ツール -> Websocketクライアントをクリックします。

  2. ユーザー名/パスワードまたは自動生成認証を使ってシミュレートクライアントとして接続します。

  3. パブリッシュセクションで以下の設定でメッセージをパブリッシュします。

    • トピック：temp_hum/emqx

    • ペイロード：

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

emqx_tables_publish

このメッセージによりルールがトリガーされ、EMQX Tablesに書き込まれます。

EMQX Tablesでデータをクエリ ​

  1. EMQX Cloudコンソールにログインします。

  2. EMQX Tablesのデプロイメントに移動します。

  3. データエクスプローラーをクリックします。

  4. 以下のSQLクエリを実行します。

    sql
    SELECT * FROM "temp_hum"

クエリ結果に新たに取り込まれたレコードが表示されるはずです。

emqx_tables_query

ルール統計の確認 ​

実行時の動作やパフォーマンスを確認するには：

  1. EMQX Enterpriseダッシュボードに戻ります。

  2. データ統合 -> ルールに移動します。

  3. 作成したルールIDをクリックします。

ルールおよび関連するEMQX Tablesアクションの実行統計（成功数、失敗数など）を確認できます。

emqx_tables_statistics