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

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

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

対象テーブルの計画

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

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

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

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

タグとフィールドの選択

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

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

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

• 少なくとも1つのフィールドカラムを含めること。
• タグは文字列値を使用してください。必要に応じて数値やブール値を文字列に変換してください。完全なpayload、オブジェクト、配列、nullはタグにできません。
• 完全な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がメッセージを受信した時刻が必要な場合	メッセージが遅延またはバッファリングされてからパブリッシュされる場合、元のイベント時刻を表しません
ペイロードのタイムスタンプ	デバイスがデータを生成した時刻が必要な場合	デバイスの時計が正確である必要があり、タイムスタンプの単位が設定した精度に合致している必要があります

ブローカー受信時刻は常にミリ秒（ms）単位です。数値のペイロードタイムスタンプの場合、推奨される時間精度がデバイスのタイムスタンプ単位と一致しているか確認してください。

例	桁数	推奨精度
1761206400	10	秒（s）
1761206400000	13	ミリ秒（ms）
1761206400000000	16	マイクロ秒（us）
1761206400000000000	19	ナノ秒（ns）

メッセージ履歴は2026-09-23T08:00:00Zや2026-09-23T16:00:00+08:00のようなRFC 3339形式の文字列も認識します。ZまたはUTCオフセットを含めてください。このソースを選択すると時間精度はデフォルトでミリ秒になりますが、必要に応じて変更可能です。

タイムスタンプソースとして選択されたフィールドはタグやフィールドから除外され、カスタム変換式は使用できません。

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

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

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

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

ルールやアクションを編集すると管理されたデータ統合が直接変更されます。編集後はテストメッセージをパブリッシュし、マッピングのメトリクスを確認し、対象テーブルをクエリして結果を検証してください。マッピングはメッセージ履歴からのみ削除し、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;

クエリ結果はローカルタイムゾーンで表示され、ミリ秒が0でない場合はミリ秒まで表示されます。マイクロ秒やナノ秒のタイムスタンプを完全に表示するには生データ表示に切り替えてください。追加のフィルター、関数、時間ウィンドウ集計についてはSQLによるデータクエリを参照してください。

関連リンク

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