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 とデータ統合の違い
Message History は、データ統合 with EMQX Tables で使われるコネクター、ルール、アクションのモデルに基づいたガイド付きワークフローです。選択した EMQX Tables のテーブルに直接書き込み、別のメッセージストアは保持しません。マッピングを作成すると、Message History は EMQX Tables コネクターを作成または再利用し、管理されたデータ統合ルールと EMQX Tables アクションを作成します。複数のマッピングで同じコネクターを共有可能です。
必要な制御レベルに応じて、Message History とデータ統合を使い分けてください。
| 検討項目 | Message History | データ統合 |
|---|---|---|
| 設定 | ガイド付きワークフローで MQTT トピック、対象テーブル、フィールドマッピング、タイムスタンプ設定を選択 | コネクター、ルール、アクションを個別に作成・設定 |
| メッセージ処理 | ペイロードフィールドと MQTT メタデータを選択し、タグ、フィールド、タイムスタンプの役割を割り当てる | カスタム Rule SQL、変換、プレースホルダー、Form または Line Protocol 設定を使用 |
| 宛先 | マッピングごとに管理された 1 つのアクションで EMQX Tables に書き込み | 複数アクションを追加、または処理済みデータを EMQX Tables や他の対応システムに送信可能 |
| データ検証 | ブローカーのデプロイメントから最近のレコードと配信メトリクスをプレビュー | コネクター、ルール、アクションを個別に監視し、EMQX Tables の保存データをクエリ可能 |
| リソース管理 | マッピング単位でパイプラインを管理。ルールとアクションはデータ統合で編集可能だが、コネクターとマッピングのライフサイクルは Message History が管理 | コネクター、ルール、アクションを独立して管理 |
| 利用環境とネットワーク | 対象は Dedicated または Dedicated Flex デプロイメントと同一ネットワーク内の EMQX Tables デプロイメント | Message History が利用できないデプロイメントやネットワーク構成で使用可能（EMQX Tables コネクターのサポートに依存） |
シンプルなテレメトリ保持と運用の簡素化には Message History を選択し、高度な処理や複数宛先、統合リソースの独立制御が必要な場合はデータ統合を選択してください。Message History で作成したルールとアクションは、より詳細な制御が必要な場合にデータ統合から編集可能です（編集制限を参照）。
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 }Tags & Fields で
payload.device_idが Tags にあり、Data Type が String に設定されていることを確認します。payload.temperatureとpayload.humidityは Fields にあり、Data Type は Float のままにします。ターゲット列名はそれぞれdevice_id、temperature、humidityのままにします。TIP
少なくとも 1 つの Field 列が必要です。タグ、フィールド、タイムスタンプ列の名前はすべて一意でなければなりません。
Timestamp Source は
Broker Received Time（ミリ秒単位）に設定したままにします。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はタグやタイムスタンプにできず、ネストされた配列は個別マッピング不可。例は列を定義し、ブローカーが受信したメッセージをサンプリングするわけではない。 - 追加フィールド: Add Tag または Add Field で
clientid、topic、qos、username、peernameなどの MQTT メタデータを追加可能。タグはフィルタリング・グルーピング用の文字列、フィールドは計測値を保存。最低1つのフィールドを保持。マッピング制限は Choose Tags and Fields を参照。 - 列名: タグ、フィールド、タイムスタンプ列の名前はすべて一意である必要あり。
- メッセージ時間: デバイスが計測値を生成した時刻を保存する場合、Unix タイムスタンプや
2026-09-23T08:00:00Zのような RFC 3339 日時文字列を例のペイロードに追加し、Timestamp Source で該当フィールドを選択。Time Precision が単位に合っているか確認（Choose a Timestamp参照）。指定しない場合は Broker Received Time を使用。 - 既存コネクター: 選択した Tables デプロイメント、データベース、ユーザーに対して Message History コネクターが既に設定されている場合、コンソールはそれを再利用。TTL とタイムスタンプ列名はウィザードで読み取り専用。これらの設定はそのコネクター経由で最初に作成されたテーブルにのみ影響。既存の対象テーブルの設定は別途確認。
テーブル設計、列の役割選択、タイムスタンプ選択、ペイロード変更対応のガイダンスは Design and Query Message History Data を参照してください。
値の変換設定
フィールドの Data Type を選択して値の保存方法を制御します。選択肢は String、Signed Integer、Unsigned Integer、Float、Boolean です。利用可能な型は例の値に依存します。整数と小数の両方を含む可能性がある計測値は Float を選択してください。タグは常に String です。
変換をカスタマイズするには：
- Data Type 横の値変換ボタンをクリック。
- Rule SQL Expression に
float(payload.temperature)のような式を入力し、対応する Data Type を選択。Functions から関数を選択も可能。 - Preview をチェックして Apply をクリック。マッピングテーブルで Data Type を後から変更するとカスタム式はリセットされます。
- Next をクリックして現在のメッセージ例でマッピングをテスト。
プレビューは基本的な str(), int(), float(), bool() の変換をサポートします。Preview unavailable は式が無効とは限らず、適用して Next でテストしてください。
テストは現在の例のみを検証し、EMQX Tables への書き込みや将来のメッセージ成功を保証しません。マッピング作成後に メッセージをパブリッシュしてテーブルをクエリ し、正しく保存されているか確認してください。
マッピングの表示と管理
Message History リストには各マッピングの ID、MQTT トピック、データベース、テーブル、有効状態が表示されます。マッピングは最終更新日時順に並びます。
- マッピング行を展開して最大10件の最近のレコードをプレビュー可能。
- マッピングの ID をクリックすると宛先、TTL、配信メトリクス、最大100件の最近のレコードを表示。
- プレビューを更新して最新レコードを読み込み。
- より広範なクエリには Data Explorer を利用。
Message History はプレビューごとに EMQX Tables をクエリし、タイムスタンプ列で結果をソートします。時間列はローカル時間と UTC オフセットを表示し、ミリ秒がゼロでない場合はミリ秒も表示。ヘッダーをクリックするとマイクロ秒やナノ秒を含む生のタイムスタンプを表示。切り捨て表示の値にカーソルを合わせると完全な内容を確認可能。
マッピングは Message History から有効化、無効化、削除できます。
- Enable 列のスイッチでマッピングを無効化または有効化可能。無効化すると新規書き込みは停止しますが、既存の EMQX Tables のデータは保持されます。
- マッピングを削除するには削除アイコンをクリックし、確認ダイアログにマッピング ID を入力します。マッピング削除は対象テーブルや保存データを削除しません。関連コネクターを他に使うマッピングがなければ、コネクターも削除されます。
以下の表は操作が保存データに与える影響をまとめています。
| 操作 | 新規書き込み | 保存データ | 対象テーブル |
|---|---|---|---|
| マッピングを無効化 | 停止 | 保持 | 保持 |
| マッピングを削除 | 停止 | 保持 | 保持 |
| データが TTL に達する | 継続 | TTL より古いデータは期限切れで削除 | 保持 |
基盤となるデータ統合の編集
マッピングの ID をクリックし、編集アイコンをクリックするとデータ統合でルールを開けます。Rule SQL、説明、アクション設定の編集や、ルール・アクションの有効化・無効化が可能です。
Message History の Enable スイッチはルールのみを制御します。データ統合で関連アクションが無効化されている場合、マッピングは有効表示でも、マッチしたメッセージは EMQX Tables に書き込まれません。
コネクターはロックされています。データ統合や Flow の Cloud Console では、シンクアクションの追加、フォールバック関係の変更、管理されたルールやアクションの削除はできません。汎用のデータ統合 API はこれらの制限をすべて強制しません。Message History のリソース削除やコネクター・フォールバック関係の変更にこれら API を使わないでください。マッピングと管理されたルール・アクションを削除するには Message History に戻って操作してください。関連コネクターは他に使うマッピングがなければ同時に削除されます。
Message History のトラブルシューティング
マッピング詳細を開き、Success、Failed、Dropped メトリクスと最近のメッセージ状態を確認してください。次に該当する項目を参照してください。
配信メトリクスが変化しない
- Message History でマッピングが有効で、データ統合でアクションが有効か確認。
- パブリッシュしたトピックがマッピングの 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 Explorer で対象テーブルをクエリ。
- Data Explorer がレコードを返す場合、最近のメッセージプレビューを更新。
- プレビューが依然として読み込めない場合は Data Explorer で保存データを検証し続ける。
既存テーブルへのマッピング書き込み
既存テーブルのタイムスタンプ列、TTL、スキーマを確認。マッピングのコネクター設定は既存テーブルを変更しません。