EMQX Cloud

日本語

简体中文
English

日本語

简体中文
English

Appearance

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

メッセージ履歴 ​

メッセージ履歴は、MQTTメッセージの選択したフィールドをEMQX Tablesに保存する機能です。MQTTトピックフィルター、ターゲットテーブル、および保存するフィールドを選択するマッピングを作成します。

EMQX Tablesはデータの保存とクエリを行いますが、EMQXブローカーのデプロイメントから自動的にメッセージを受信するわけではありません。メッセージ履歴がない場合は、データ統合コネクター、ルール、アクションを手動で設定する必要があります。メッセージ履歴は、この取り込みパイプラインを3ステップのマッピングワークフローで設定および管理します。

メッセージ履歴を使うと、以下が可能です。

  • データ統合パイプラインを手動で構築せずにMQTTデータの保存を設定できる。
  • ブローカーのデプロイメントから最近のレコードをプレビューし、配信メトリクスを確認できる。
  • EMQX Tablesにテレメトリを保持し、SQLクエリやトレンド分析に活用できる。
  • 個別のマッピングを有効化、無効化、削除でき、既にEMQX Tablesに保存されたデータは削除されない。

メッセージ履歴は、EMQX v6.1.4以降を実行するDedicatedおよびDedicated Flexの国際サイトでのみ利用可能です。

メッセージ履歴の仕組み ​

マッピングウィザードは3つのステップで構成されています。

  • 保存場所：ブローカーと同じネットワーク内のEMQX Tablesデプロイメントとデータベースを選択します。Tablesユーザーを選び、そのパスワードを入力します。次に保持期間（TTL）とタイムスタンプ列名を設定します。
  • メッセージとテーブルスキーマ：MQTTトピックフィルターとターゲットテーブルを入力し、メッセージの値をタグまたはフィールド列にマッピングし、タイムスタンプのソースと精度を選択します。
  • 確認：保存場所、トピック、テーブルスキーマを確認し、マッピングを作成します。

保存場所を選択すると、メッセージ履歴はEMQX Tablesへの接続を作成または再利用します。マッピングを確定すると、データ統合ルールとEMQX Tablesアクションが作成されます。ルールはトピックフィルターに一致するメッセージから値を選択し、アクションはそれらの値をターゲットテーブルに書き込みます。

ターゲットテーブルが存在しない場合、最初の成功した書き込みでTTLとタイムスタンプ列名を指定してテーブルが作成されます。既存のテーブルにはこれらの設定は適用されません。最初の成功書き込みの値が列の型を決定するため、以降のメッセージは互換性のある型を使用する必要があります。メッセージ例はスキーマのプレビューのみであり、テーブル作成やレコード書き込みは行いません。

マッピングが有効な間に受信したメッセージのみが保存されます。メッセージ履歴は過去のMQTTトラフィックのバックフィルや保存済みメッセージのサブスクライバーへの再生は行いません。クライアント接続やサブスクリプションイベントについては、イベント履歴を使用してください。

一致する各メッセージはデータ統合のTPSを消費します。マッピング作成前にデプロイメントのデータ統合TPS容量を確認してください。

はじめる前に ​

  • ブローカーのデプロイメントが稼働していることを確認してください。
  • ブローカーと同じネットワーク内で稼働中のEMQX Tablesデプロイメントを用意してください。NATゲートウェイやPrivateLink経由で別ネットワークからアクセスするTablesデプロイメントは対象外です。
  • EMQX Tablesのデータベースとアクセス権限を持つユーザーを用意してください。ユーザーにはSqlSelectとSqlInsert権限が必要です。Broker Integrationプリセットは両方を付与します。ユーザーのパスワードを準備してください。詳細はEMQX Tablesユーザー管理を参照してください。

例：センサーの読み取り値を保存する ​

この例では、デバイスがsensors/sensor-001/telemetryのようなトピックに温度と湿度の読み取り値をパブリッシュします。マッピングはpublicデータベースのsensor_historyテーブルにすべてのセンサーの読み取り値を保存します。タイムスタンプ列にはブローカー受信時間を使用するため、ペイロードにタイムスタンプは不要です。

マッピングの作成 ​

  1. クラウドコンソールでブローカーのデプロイメントを開き、左メニューからメッセージ履歴を選択します。初めての場合はトピックをテーブルにマッピングを、既にマッピングがある場合は新規マッピングをクリックします。

  2. 保存場所でEMQX Tablesデプロイメント、publicデータベース、Tablesユーザーを選択し、ユーザーのパスワードを入力します。**Time-To-Live (TTL)**は30 days、タイムスタンプ列名はtimestampのままにします。次へをクリックして接続を作成または再利用します。

    保存場所の設定

  3. メッセージとテーブルスキーマで、MQTTトピックにsensors/+/telemetry、テーブル名にsensor_historyを入力します。+ワイルドカードは1つのトピックレベルにマッチするため、異なるセンサーIDの読み取り値を含みます。

  4. MQTTメッセージ例を以下のペイロードに置き換えます。

    json
    {
  "device_id": "sensor-001",
  "temperature": 23.5,
  "humidity": 58.5
}

  5. フィールドマッピングでpayload.device_idをタグに設定します。payload.temperatureとpayload.humidityはフィールドのままにし、ターゲット列名はそれぞれdevice_id、temperature、humidityのままにします。

    TIP

    最初の成功した書き込み後、テーブルにはtimestampというタイムスタンプ列、デバイスでフィルタリングするためのdevice_idタグ、読み取り値用のtemperatureとhumidityフィールドが作成されます。フィールド列は最低1つ必要で、ターゲット列名はユニークでなければなりません。

  6. タイムスタンプ列はブローカー受信時間、時間精度は**ミリ秒 (ms)**のままにして、次へをクリックします。

    メッセージとテーブルスキーマの設定

  7. 保存先、トピック、スキーマプレビューを確認し、確定をクリックします。マッピングがメッセージ履歴リストに表示されます。

    マッピング設定の確認

メッセージのパブリッシュとクエリ ​

  1. ブローカーのデプロイメントで診断 -> MQTTクライアントを開き、接続して例のペイロードをsensors/sensor-001/telemetryにパブリッシュします。

  2. メッセージ履歴に戻り、マッピング行を展開して最近のレコードをプレビューします。マッピングのIDをクリックして詳細と配信メトリクスを確認します。

  3. リンクされたEMQX Tablesデプロイメントのデータエクスプローラーを開き、publicデータベースを選択して以下を実行します。

    sql
    SELECT * FROM sensor_history ORDER BY "timestamp" DESC LIMIT 10;

クエリは、パブリッシュされた温度と湿度を持つsensor-001の行を返します。詳細なクエリオプションはデータエクスプローラーを参照してください。

マッピング詳細とメトリクスの表示

他のメッセージ形式の設定 ​

  • トピックフィルター：MQTTの+および#ワイルドカードを使って複数トピックをキャプチャできます。マッピングは有効時に受信したメッセージのみ保存し、過去のメッセージはコピーしません。
  • メッセージ例：コンソールはJSONオブジェクトのフィールドとネストされたパスを検出します。プレーンテキストペイロードは1つのpayloadフィールドになります。配列はサポートされません。例は列を定義し、ブローカーが受信したメッセージのサンプリングではありません。
  • 追加フィールド：フィールド追加でclientid、topic、qos、username、peernameなどのMQTTメタデータをマッピングできます。フィルタやグループ化に使う値はタグ、測定値はフィールドを選択します。フィールド列は最低1つ必要です。
  • メッセージ時間：デバイスが読み取り値を生成した時刻を保存するには、Unixタイムスタンプや2026-09-23T08:00:00Zのような日時文字列を例のペイロードに追加し、タイムスタンプ列でそのフィールドを選択します。Unixタイムスタンプの場合は時間精度を単位に合わせて設定してください。コンソールは認識したRFC 3339文字列を選択した精度に変換します。指定がなければブローカー受信時間を使用します。
  • 既存の接続：既に選択したTablesデプロイメント、データベース、ユーザーを使うメッセージ履歴接続がある場合、コンソールはそれを再利用します。TTLとタイムスタンプ列名はウィザードで読み取り専用です。これらの設定はその接続で最初に作成されたテーブルにのみ影響します。既存のターゲットテーブルの設定は別途確認してください。

設定と書き込みのトラブルシューティング ​

  • 次へをクリックしても接続が作成されない場合、Tablesデプロイメントがブローカーと同じネットワークで稼働中か確認してください。Tablesの認証情報とデータベースアクセスもチェックしてください。メッセージ履歴接続はブローカーのデータ統合コネクターの上限を共有します。
  • レコードが表示されない場合、マッピングが有効であり、パブリッシュされたトピックがMQTTトピックフィルターに一致しているか確認してください。マッピング作成前のメッセージはテーブルにコピーされません。
  • マッピング詳細の成功、失敗、破棄メトリクスを確認してください。Tablesデプロイメントが利用不可、データベースアクセス不足、タイムスタンプ無効、値がテーブルの列型と不一致などで書き込みが失敗することがあります。
  • 既存テーブル用にマッピングを作成した場合は、EMQX Tablesでそのテーブルのタイムスタンプ列、TTL、スキーマを確認してください。マッピングの接続設定は既存テーブルを変更しません。

マッピングの表示と管理 ​

メッセージ履歴リストには、各マッピングのID、MQTTトピック、データベース、テーブル、有効状態が表示されます。

  • マッピング行を展開して最大10件の最近のレコードをプレビューできます。
  • マッピングのIDをクリックすると、保存先、TTL、配信メトリクス、最大100件の最近のレコードを表示できます。
  • プレビューを更新して最新のレコードを読み込みます。
  • より広範なクエリにはデータエクスプローラーを使用します。

メッセージ履歴は各プレビューでEMQX Tablesにクエリを送り、タイムスタンプ列で結果をソートします。

最近のメッセージをプレビュー

マッピングは作成後に編集できません。以下の操作が可能です。

  • 有効化列のスイッチでマッピングを無効化または有効化できます。無効化すると新規書き込みは停止しますが、既存のEMQX Tablesのデータは保持されます。
  • マッピングを削除するには、削除アイコンをクリックし、確認ダイアログにマッピングのIDを入力します。マッピングを削除してもターゲットテーブルや保存データは削除されません。専用接続を使用している他のマッピングがなければ、その接続も削除されます。