Message History
Message History は、MQTT メッセージの選択したフィールドを EMQX Tables に保存できる機能です。MQTT トピックフィルター、ターゲットテーブル、保存するフィールド、および各レコードのタイムスタンプを選択するマッピングを作成します。
Message History を使うと、以下が可能です。
- 手動でデータ統合パイプラインを構築せずに MQTT データの保存を設定できる。
- ブローカーのデプロイメントから最近のレコードをプレビューし、配信メトリクスを確認できる。
- テレメトリを EMQX Tables に保持し、SQL クエリやトレンド分析に利用できる。
- 個別のマッピングを有効化、無効化、削除でき、既に EMQX Tables に保存されたデータは削除されない。
Message History は、EMQX v6.1.4 以降を実行する Dedicated および Dedicated Flex の国際サイト限定のデプロイメントで利用可能です。
Message History を使うタイミング
MQTT メッセージの選択したフィールドを継続的に EMQX Tables に保存したい場合に Message History を使用します。デバイスの測定値、ステータスレポート、環境データなど、構造が一貫したテレメトリに適しています。
Message History と Data Integration の違い
Message History は、Data Integration with EMQX Tables で使われるコネクター、ルール、アクションモデルを基にしたガイド付きワークフローです。選択した EMQX Tables テーブルに直接書き込み、別のメッセージストアは保持しません。マッピング作成時に、Message History は EMQX Tables コネクターを作成または再利用し、管理されたデータ統合ルールと EMQX Tables アクションを作成します。複数のマッピングが同じコネクターを共有可能です。
必要な制御レベルに応じて、Message History と Data Integration を使い分けてください。
| 比較項目 | Message History | Data Integration |
|---|---|---|
| セットアップ | ガイド付きワークフローで MQTT トピック、ターゲットテーブル、フィールドマッピング、タイムスタンプ設定を選択 | コネクター、ルール、アクションを個別に作成・設定 |
| メッセージ処理 | ペイロードフィールドと MQTT メタデータを選択し、Tag、Field、タイムスタンプの役割を割り当て | カスタム Rule SQL、変換、プレースホルダー、Form または Line Protocol 設定を利用 |
| 書き込み先 | マッピングごとに管理された 1 つのアクションで EMQX Tables に書き込み | 複数アクションの追加や EMQX Tables 以外のシステムへの送信も可能 |
| データ検証 | ブローカーのデプロイメントから最近のレコードと配信メトリクスをプレビュー | コネクター、ルール、アクションを個別に監視し、EMQX Tables からデータをクエリ |
| リソース管理 | マッピング単位でパイプライン管理。Data Integration でルール・アクション設定を編集可能だが、コネクターとマッピングのライフサイクルは Message History が管理 | コネクター、ルール、アクションを独立して管理 |
| 利用可能性とネットワーク | 対象の Dedicated または Dedicated Flex デプロイメントと同一ネットワーク内の EMQX Tables デプロイメントで利用可能 | Message History が利用できないデプロイメントやネットワーク構成で利用可能（EMQX Tables コネクターの対応状況による） |
シンプルなテレメトリ保持と運用の簡素化を求める場合は Message History を選択し、高度な処理や複数の送信先、統合リソースの独立制御が必要な場合は Data Integration を選択してください。Message History で作成したルールとアクションは、より詳細な制御が必要な場合に Data Integration で編集可能です。ただし、編集制限に注意してください。
Message History はマッピングが有効な間に受信したメッセージのみを保存し、過去の MQTT トラフィックを遡って保存したり、保存済みメッセージをサブスクライバーに再送信したりしません。マッチしたメッセージはデータ統合の TPS を消費し、Message History のコネクターはブローカーのコネクター制限にカウントされます。マッピング作成前にデプロイメントのデータ統合 TPS容量を確認してください。
クライアント接続、認証、サブスクリプションイベントの調査にはEvent Historyを、EMQX Tables に保存されたデータに対する広範なクエリにはData Explorerを使用してください。
はじめる前に
- ブローカーのデプロイメントが稼働中で、設定権限があることを確認してください。
- ブローカーと同じネットワーク内で稼働中の EMQX Tables デプロイメントを用意してください。NAT ゲートウェイや PrivateLink 経由で別ネットワークからアクセスする Tables デプロイメントは対象外です。対象の Tables デプロイメントがない場合、Message History はブローカーのプロジェクト、プラットフォーム、リージョン、ネットワークに新規作成するための New EMQX Tables エントリを提供します。
- EMQX Tables のデータベースとアクセス権を持つユーザーを用意してください。ユーザーには
SqlSelectとSqlInsert権限が必要で、Broker Integration プリセットが両方付与します。ユーザーのパスワードも準備してください。EMQX Tables ユーザー管理を参照してください。
センサーの読み取り値を EMQX Tables に保存する
この例では、デバイスが sensors/sensor-001/telemetry のようなトピックに温度と湿度の読み取り値をパブリッシュします。マッピングはすべてのセンサーの読み取り値を public データベースの sensor_history テーブルに保存します。タイムスタンプ列にはブローカー受信時刻を使うため、ペイロードにタイムスタンプは不要です。
マッピングの作成
Cloud Console でブローカーのデプロイメントを開き、左メニューから Message History を選択します。最初のマッピングの場合は Map a Topic to a Table、既存マッピングがある場合は New Mapping をクリックします。
Storage Location で EMQX Tables デプロイメント、
publicデータベース、Tables ユーザーを選択し、ユーザーのパスワードを入力します。Time-To-Live (TTL) は30 days、タイムスタンプ列名はtimestampのままにします。Next をクリックして新しい接続をテストするか既存接続を再利用します。接続テストは作成を伴いません。
Message & Table Schema で MQTT Topic に
sensors/+/telemetry、Table Name にsensor_historyを入力します。+ワイルドカードはトピックレベル1つにマッチするため、異なるセンサーIDの読み取り値を含みます。MQTT Message Example を以下のペイロードに置き換えます。
json{ "device_id": "sensor-001", "temperature": 23.5, "humidity": 58.5 }Field Mapping で
payload.device_idを Tag に設定し、payload.temperatureとpayload.humidityは Field のままにします。ターゲット列名はそれぞれdevice_id、temperature、humidityのままにします。TIP
Field 列は最低1つ必要です。Tag、Field、タイムスタンプ列の名前はすべて一意でなければなりません。
Timestamp Column は
Broker Received Time、Time Precision はMilliseconds (ms)のままにして Next をクリックします。
書き込み先、トピック、スキーマプレビューを確認し、Confirm をクリックします。Message History は必要に応じてコネクターを作成し、続いてデータ統合ルールと EMQX Tables アクションを作成します。マッピングが Message History リストに表示されます。
スキーマプレビューはテーブル作成やレコード書き込みを行いません。ターゲットテーブルが存在しない場合、最初の正常な書き込み時に TTL とタイムスタンプ列名で作成されます。コネクター作成は成功したがマッピング作成が失敗した場合、コネクターは保持され再試行時に再利用されます。保持されたコネクターはブローカーのコネクター制限にカウントされ続けます。
メッセージをパブリッシュして確認する
ブローカーのデプロイメントで Diagnostics -> MQTT Client を開き、接続して例のペイロードを
sensors/sensor-001/telemetryにパブリッシュします。Message History に戻り、マッピング行を展開して最近のレコードをプレビューします。マッピングの ID をクリックして詳細を開き、配信メトリクスを確認します。
連携している EMQX Tables デプロイメントの Data Explorer を開き、
publicデータベースを選択して以下を実行します。sqlSELECT * FROM sensor_history ORDER BY "timestamp" DESC LIMIT 10;
クエリは sensor-001 の行を返し、パブリッシュした温度と湿度を含みます。その他の例は Query Stored Messages を参照してください。
メッセージフォーマットとフィールドマッピングの設定
- トピックフィルター: MQTT の
+と#ワイルドカードを使って複数トピックをキャプチャ可能。マッピングは有効時に受信したメッセージのみ保存します。 - メッセージ例: コンソールは JSON オブジェクトのフィールドとネストされたパスを検出します。プレーンテキストは単一の
payloadフィールドとなり、トップレベル配列は単一の文字列フィールドとして保存可能です。完全なpayloadは Tag またはタイムスタンプにできず、ネストされた配列も個別にマッピングできません。例は列を定義し、ブローカーが受信したメッセージのサンプルではありません。 - 追加フィールド: Add Field で
clientid、topic、qos、username、peernameなどの MQTT メタデータをマッピング可能。フィルターやグループ化に使うテキスト値は Tag、測定値は Field を選択。数値は Tag にできません。最低1つの Field 列を保持してください。 - 列名: Tag、Field、タイムスタンプ列の名前はすべて一意でなければなりません。
- メッセージ時刻: デバイスの読み取り時刻を保存するには、Unix タイムスタンプや
2026-09-23T08:00:00Zのような RFC 3339 日時文字列を例のペイロードに追加し、Timestamp Column でそのフィールドを選択します。Unix タイムスタンプの場合は単位に合わせて Time Precision を設定してください。そうでなければ Broker Received Time を使用します。 - 既存コネクター: 選択した Tables デプロイメント、データベース、ユーザーに Message History コネクターが既にある場合、コンソールはそれを再利用します。TTL とタイムスタンプ列名はウィザードで読み取り専用です。これらの設定はそのコネクターで最初に作成されたテーブルにのみ影響します。既存のターゲットテーブルの設定は別途確認してください。
テーブル設計、列の役割選択、タイムスタンプの選択、ペイロード変更の扱いについては Design and Query Message History Data を参照してください。
マッピングの表示と管理
Message History リストには各マッピングの ID、MQTT トピック、データベース、テーブル、有効状態が表示されます。マッピングは最終更新日時順に並びます。
- マッピング行を展開して最大10件の最近のレコードをプレビュー可能。
- マッピングの ID をクリックすると、書き込み先、TTL、配信メトリクス、最大100件の最近のレコードを表示。
- プレビューを更新して最新レコードを読み込み。
- より広範なクエリには Data Explorer を使用。
Message History は各プレビューで EMQX Tables にクエリを送り、タイムスタンプ列で結果をソートします。時間列はローカル時刻と UTC オフセットを表示し、見出しをクリックすると生のタイムスタンプに切り替わります。省略された値にカーソルを合わせると全内容を確認できます。
Message History でマッピングの有効化、無効化、削除が可能です。
- Enable 列のスイッチでマッピングを無効化または有効化できます。無効化すると新規書き込みは停止しますが、既存の EMQX Tables データは保持されます。
- マッピングを削除するには削除アイコンをクリックし、確認ダイアログでマッピング ID を入力します。マッピング削除はターゲットテーブルや保存済みデータを削除しません。関連コネクターを他のマッピングが使っていなければ、コネクターも削除されます。
以下の表はこれらの操作が保存データに与える影響をまとめたものです。
| 操作 | 新規書き込み | 保存データ | ターゲットテーブル |
|---|---|---|---|
| マッピングを無効化 | 停止 | 保持 | 保持 |
| マッピングを削除 | 停止 | 保持 | 保持 |
| データが TTL に達した場合 | 継続 | TTL より古いデータは期限切れ | 保持 |
基盤となる Data Integration の編集
マッピングの ID をクリックし、編集アイコンをクリックすると Data Integration で対応するルールを開けます。Rule SQL、説明、アクション設定の編集や、ルール・アクションの有効化・無効化が可能です。
Message History の Enable スイッチはルールのみを制御します。関連アクションが Data Integration で無効化されている場合、マッピングは有効表示されても、マッチしたメッセージは EMQX Tables に書き込まれません。
コネクターはロックされています。Data Integration と Flow の Cloud Console では、シンクアクションの追加、フォールバック関係の変更、管理されたルールやアクションの削除はできません。一般的な Data Integration API はこれらの制限をすべて強制しません。Message History のリソース削除やコネクター・フォールバック関係の変更にこれらの API を使わないでください。マッピングと管理されたルール・アクションを削除するには Message History に戻って操作してください。関連コネクターは他のマッピングが使っていなければ同時に削除されます。
Message History のトラブルシューティング
マッピングの詳細を開き、Success、Failed、Dropped のメトリクスと最近のメッセージの状態を確認してください。観察された結果に応じて以下のセクションを参照してください。
配信メトリクスが変化しない
- Message History でマッピングが有効で、Data Integration で対応するアクションが有効か確認してください。
- パブリッシュしたトピックがマッピングの MQTT トピックフィルターに合致しているか確認してください。
- 新しいマッチするメッセージをパブリッシュしてください。マッピング作成前や無効化中にパブリッシュされたメッセージはテーブルにコピーされません。
Failed または Dropped が増加する
- Diagnostics -> Logs を開き、Error Type を Data Integration に設定し、マッピングの Rule ID、Action Name、または Connector Name でフィルターしてエラーを特定してください。フィルターの詳細は Deployment Logs を参照。
- Tables デプロイメントがブローカーと同じネットワークで稼働していることを確認してください。
- Tables の認証情報が正しく、ユーザーにデータベースへの
SqlSelectとSqlInsert権限があることを確認してください。 - マッピングされた値がターゲットテーブルのスキーマと合致しているか確認してください。値が既存の列の型と異なる場合やタイムスタンプが無効な場合、書き込みに失敗します。
- ペイロードのタイムスタンプが選択した精度に合っているか確認してください。例えば秒単位のタイムスタンプをミリ秒として設定しないでください。
- デプロイメントのデータ統合 TPS とコネクター制限を確認してください。Message History のコネクターは Data Integration のコネクターとブローカーの制限を共有します。
最近のメッセージが表示されない
- ターゲットテーブルが準備できていない場合、マッチするメッセージをパブリッシュして最初の正常な書き込みをトリガーしてください。
- Data Explorer からターゲットテーブルをクエリしてください。
- Data Explorer でレコードが返る場合、最近のメッセージプレビューを更新してください。
- プレビューが依然として読み込めない場合は、Data Explorer を使って保存データを検証し続けてください。
既存テーブルへのマッピング書き込み
既存テーブルのタイムスタンプ列、TTL、スキーマを確認してください。マッピングのコネクター設定は既存テーブルを変更しません。