# メッセージ履歴 メッセージ履歴は、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トラフィックのバックフィルや保存済みメッセージのサブスクライバーへの再生は行いません。クライアント接続やサブスクリプションイベントについては、[イベント履歴](./event_history.md)を使用してください。 一致する各メッセージはデータ統合のTPSを消費します。マッピング作成前にデプロイメントの[データ統合TPS](./metrics.md)容量を確認してください。 ## はじめる前に - ブローカーのデプロイメントが稼働していることを確認してください。 - ブローカーと同じネットワーク内で稼働中のEMQX Tablesデプロイメントを用意してください。NATゲートウェイやPrivateLink経由で別ネットワークからアクセスするTablesデプロイメントは対象外です。 - EMQX Tablesのデータベースとアクセス権限を持つユーザーを用意してください。ユーザーには`SqlSelect`と`SqlInsert`権限が必要です。**Broker Integration**プリセットは両方を付与します。ユーザーのパスワードを準備してください。詳細は[EMQX Tablesユーザー管理](../emqx_tables/emqx_tables_user_management.md)を参照してください。 ## 例：センサーの読み取り値を保存する この例では、デバイスが`sensors/sensor-001/telemetry`のようなトピックに温度と湿度の読み取り値をパブリッシュします。マッピングは`public`データベースの`sensor_history`テーブルにすべてのセンサーの読み取り値を保存します。タイムスタンプ列にはブローカー受信時間を使用するため、ペイロードにタイムスタンプは不要です。 ### マッピングの作成 1. クラウドコンソールでブローカーのデプロイメントを開き、左メニューから**メッセージ履歴**を選択します。初めての場合は**トピックをテーブルにマッピング**を、既にマッピングがある場合は**新規マッピング**をクリックします。 2. **保存場所**でEMQX Tablesデプロイメント、`public`データベース、Tablesユーザーを選択し、ユーザーのパスワードを入力します。**Time-To-Live (TTL)**は`30 days`、タイムスタンプ列名は`timestamp`のままにします。**次へ**をクリックして接続を作成または再利用します。 ![保存場所の設定](./_assets/message-history-storage-location.png) 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)**のままにして、**次へ**をクリックします。 ![メッセージとテーブルスキーマの設定](./_assets/message-history-message-schema.png) 7. 保存先、トピック、スキーマプレビューを確認し、**確定**をクリックします。マッピングが**メッセージ履歴**リストに表示されます。 ![マッピング設定の確認](./_assets/message-history-confirmation.png) ### メッセージのパブリッシュとクエリ 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`の行を返します。詳細なクエリオプションは[データエクスプローラー](../emqx_tables/emqx_tables_data_explorer.md)を参照してください。 ![マッピング詳細とメトリクスの表示](./_assets/message-history-mapping-details.png) ## 他のメッセージ形式の設定 - **トピックフィルター**：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にクエリを送り、タイムスタンプ列で結果をソートします。 ![最近のメッセージをプレビュー](./_assets/message-history-recent-messages.png) マッピングは作成後に編集できません。以下の操作が可能です。 - **有効化**列のスイッチでマッピングを無効化または有効化できます。無効化すると新規書き込みは停止しますが、既存のEMQX Tablesのデータは保持されます。 - マッピングを削除するには、削除アイコンをクリックし、確認ダイアログにマッピングの**ID**を入力します。マッピングを削除してもターゲットテーブルや保存データは削除されません。専用接続を使用している他のマッピングがなければ、その接続も削除されます。