メッセージ履歴データの設計とクエリ

このガイドでは、メッセージ履歴マッピングのテーブル構造、カラムの役割、タイムスタンプの計画方法について説明します。また、EMQX Tablesに保存されたメッセージをクエリするためのSQL例も提供しています。

マッピングの作成手順については、メッセージ履歴を参照してください。

対象テーブルの計画

メッセージ履歴マッピングは、MQTTトピックフィルターに一致するメッセージから値を選択し、1つのEMQX Tablesテーブルに書き込みます。テーブルは、同じデバイスタイプやアプリケーションからの読み取り値など、一貫した構造を持つメッセージを中心に計画してください。

テーブルを計画する際は以下を考慮してください。

• メッセージタイプのフィールドや値の型が互換性がない場合は、別のテーブルを使用してください。
• クエリ、フィルター、グループ化、分析に必要な値のみをマッピングしてください。マッピングされていないペイロードのメンバーは、完全なpayloadを単一のフィールドとしてマッピングしない限り、別のテーブルカラムを作成しません。
• 各マッピングされたカラムの値の型は、デバイスやメッセージ間で一貫性を保ってください。例えば、あるメッセージで数値のtemperatureをパブリッシュし、別のメッセージで文字列としてパブリッシュしないでください。
• 既存のテーブルに書き込む前に、タイムスタンプカラム、TTL、カラム型を確認してください。マッピングは既存テーブルのこれらの設定を変更しません。

対象テーブルが存在しない場合、最初の正常な書き込み時にテーブルが作成されます。その書き込みの値がカラム型を決定し、以降の書き込みは互換性のある型を使用する必要があります。マッピングウィザードのMQTTメッセージ例はスキーマのプレビューのみであり、テーブルの作成や変更は行いません。

タグとフィールドの選択

タグは時系列を識別または分類するテキスト値に使用し、フィールドは測定値や時間とともに変化する値に使用します。

データ	推奨される役割	例
レコードのフィルターやグループ化に使うテキスト	タグ	device_id、region、model
測定値や変化する値	フィールド	temperature、humidity、voltage
レコードの時刻	タイムスタンプ	ブローカー受信時刻またはデバイスのタイムスタンプ

以下のマッピングルールを適用してください。

• 少なくとも1つのフィールドカラムを含めること。
• 数値をタグとして設定しないこと。
• 完全なpayloadはフィールドとしてのみ保存すること。
• 最上位の配列は1つの文字列フィールドとして保存可能。ネストされた配列は個別にマッピングできません。
• タグ、フィールド、タイムスタンプカラムはそれぞれ固有の名前を使用してください。

例として、以下のペイロードを考えます。

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

適切なマッピング例は以下の通りです。

ソースフィールド	対象カラム	役割
payload.device_id	device_id	タグ
payload.region	region	タグ
payload.temperature	temperature	フィールド
payload.humidity	humidity	フィールド

この構成により、デバイスや地域でレコードをフィルターやグループ化しつつ、測定値から統計を計算できます。

タイムスタンプの選択

すべてのメッセージ履歴レコードにはタイムスタンプが必要です。レコードがEMQXがメッセージを受信した時刻を表すべきか、デバイスがデータを生成した時刻を表すべきかに応じて、タイムスタンプのソースを選択してください。

タイムスタンプのソース	使用する場合	注意点
ブローカー受信時刻	デバイスが信頼できるタイムスタンプを提供しない場合、またはEMQXがメッセージを受信した時刻が必要な場合	メッセージが遅延またはバッファリングされてからパブリッシュされた場合、元のイベント時刻を表しません
ペイロードのタイムスタンプ	デバイスがデータを生成した時刻が必要な場合	デバイスの時計が正確である必要があり、タイムスタンプの単位は設定された精度と一致する必要があります

Unixタイムスタンプの場合、値に合った精度を選択してください。

例の値	精度
1761206400	秒（s）
1761206400000	ミリ秒（ms）
1761206400000000	マイクロ秒（us）
1761206400000000000	ナノ秒（ns）

メッセージ履歴は、2026-09-23T08:00:00Zや2026-09-23T16:00:00+08:00のようなRFC 3339形式の文字列も認識します。タイムスタンプが想定されるタイムゾーンに依存しないように、Zまたは明示的なUTCオフセットを含めてください。

デバイスがオフライン中にメッセージをバッファリングできる場合、ペイロードのタイムスタンプは元の測定時刻を保持できます。一方、ブローカー受信時刻はバッファされたメッセージが最終的にEMQXに到達した時刻を記録します。

ペイロードおよびスキーマの変更への対応

デバイスのペイロードが変更された場合は、マッピングを見直してください。以下のガイドラインを参考にしてください。

• マッピングされていないフィールドがペイロードに追加されても、マッピングが完全なpayloadを単一フィールドとして保存していない限り、メッセージ履歴は別のテーブルカラムを作成しません。
• 既存のマッピングに新しいフィールドを保存するには、ルールSQLとData Integrationの対応するEMQX Tablesアクションの書き込み設定の両方を更新してください。あるいは、新しいマッピングを作成してください。
• マッピングされたフィールドの型が変更された場合、既存テーブルのカラム型が維持されるため、書き込みが失敗する可能性があります。
• すべてのメッセージにフィールドが存在しなくなった場合は、全デバイスに展開する前に変更後のペイロードをテストしてください。
• 新しい構造が既存テーブルと互換性がない場合は、互換性のないレコードを混在させるのではなく、新しいマッピングと対象テーブルを作成してください。

ルールやアクションを編集すると、管理されたデータ統合が直接変更されます。編集後はテストメッセージをパブリッシュし、マッピングのメトリクスを確認し、対象テーブルをクエリして結果を検証してください。マッピングはメッセージ履歴からのみ削除し、EMQX Cloudが管理リソースをクリーンアップできるようにしてください。

保存されたメッセージのクエリ

リンクされたEMQX TablesのデプロイメントでData Explorerを開き、マッピングのデータベースを選択します。以下の例では、publicデータベースのsensor_historyテーブルと、timestampという名前のタイムスタンプカラムを使用しています。

最新メッセージのクエリ

最新の10件のレコードを返します。

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

特定デバイスのメッセージのクエリ

device_idタグでレコードをフィルターします。

SELECT "timestamp", temperature, humidity
FROM public."sensor_history"
WHERE device_id = 'sensor-001'
ORDER BY "timestamp" DESC;

時間範囲のクエリ

直近5分間のタイムスタンプに該当するレコードを返します。

SELECT *
FROM public."sensor_history"
WHERE "timestamp" >= now() - '5 minutes'::INTERVAL
ORDER BY "timestamp" DESC;

測定値の集計

直近1時間の各デバイスごとの平均温度を計算します。

SELECT device_id, avg(temperature) AS avg_temperature
FROM public."sensor_history"
WHERE "timestamp" >= now() - '1 hour'::INTERVAL
GROUP BY device_id;

EMQX Tablesは、Data Explorerクライアントのタイムゾーンに従ってタイムスタンプ結果をフォーマットします。追加のフィルター、関数、時間ウィンドウ集計については、SQLによるデータクエリを参照してください。

関連リンク

• メッセージ履歴
• Data Explorerでのデータクエリ
• SQLによるデータクエリ
• EMQX Tablesユーザー管理