Quick Start: Ingest MQTT Data into EMQX Tables

このガイドでは、EMQX ブローカーと EMQX Tables を使用して、外部データベースに依存せずに MQTT からデータベースへのパイプラインを構築する方法を説明します。EMQX Tables は、EMQX Cloud プラットフォームにネイティブに組み込まれた完全マネージドの時系列データベースであり、リアルタイムの IoT データ処理に最適化されています。

本ガイドで学べる内容：

EMQX ブローカーおよび EMQX Tables のデプロイ作成
EMQX Tables 用のカスタムユーザー作成
データ統合を使った MQTT データの EMQX Tables への取り込み
組み込みの Data Explorer を用いた SQL による時系列データのクエリ
必要に応じたカスタムデータベースおよびテーブルの作成

注意事項

本ドキュメントのスクリーンショットは参考用です。

EMQX Cloud は継続的に進化しており、コンソール UI は随時更新・改善されるため、一部のスクリーンショットが最新のインターフェースと完全に一致しない場合があります。ただし、全体のワークフローや機能は一貫しています。

EMQX Tables 無料トライアル

EMQX Tables では、ネイティブな MQTT からデータベースへの取り込みと時系列分析を無料で評価できるトライアルを提供しています。

クォータと期間

EMQX Tables 無料トライアルには以下が含まれます：

14日間の無料トライアル期間
100 GB のアウトバウンドトラフィック
100 GB のストレージ容量

有効期限

3日間連続でアクティブな接続がないトライアルデプロイは自動的に停止されます。
トライアル期間終了時にデプロイは即時停止します。
有効期限切れから3日後にインスタンスは削除されます。
削除後、すべてのデータは完全に消去されます。

トライアル終了後も EMQX Tables を継続利用するには、トライアル期間内に有料プランへアップグレードしてください。

ユースケース：スマートファクトリーモニタリング

具体例として、工場のデバイスが定期的に以下のような時系列テレメトリデータを報告するケースを考えます：

machine_id：デバイス識別子
production_line：所属する生産ライン
temperature：温度計測値
vibration：振動強度
machine_status：稼働状態（例：稼働中、警告、エラー）
ts：計測のタイムスタンプ

このデータを MQTT 経由で取り込み、EMQX Tables にネイティブに保存して監視、分析、アラートに活用します。

注意事項

サーバレスデプロイでも EMQX Tables へのデータ取り込みが可能です。本ガイドの手順は Dedicated Flex を例にしています。サーバレスの場合、データ統合の手順は同じですが、EMQX Tables への接続は TLS を用いたパブリックインターネット経由となり、ネットワークアソシエーションや NAT ゲートウェイは適用されません。

EMQX ブローカーと EMQX Tables のデプロイ作成

EMQX Cloud コンソールにログインします。
新規または既存のプロジェクトを作成または選択します。
EMQX Brokers の下で + New Deployment をクリックします。
Dedicated Flex プランを選択し、仕様を設定します。
- 必要に応じて Cloud Provider & Region を選択します。
- 以下のオプションはデフォルトのままで問題ありません（クイックデモ用）。
  - Tier
  - Subscribe to Smart Data Hub（任意）
  - Deployment Name & Project
  - EMQX Version
右下の Deploy をクリックします。
詳細は Dedicated Flex デプロイの作成を参照してください。
EMQX Tables の下で + New Deployment をクリックします。
デフォルトの Starter プランを選択し、Broker デプロイと同じクラウドプロバイダーとリージョンを選択します。
（任意）Network Association で既存のネットワークをドロップダウンから選択します。両デプロイが同じクラウドプラットフォームかつ同一リージョンであれば、Broker のネットワークがリストに表示されます。これを選択すると、両デプロイが同じネットワークを共有し、プライベート接続で通信可能になります。
Tier を選択します。
（任意）Deployment Name を入力します。
Deploy をクリックします。詳細は EMQX Tables デプロイの作成を参照してください。

new_table_deployment

デプロイが作成されたら、プロジェクト内のデプロイカードをクリックしてデプロイに入ります。

EMQX Tables のデプロイに入り、左メニューの Data Explorer をクリックすると、デフォルトの public データベースが利用可能であることが確認できます。

データ取り込み用ユーザーの作成

新しい EMQX Tables デプロイでは、カスタムユーザーと権限設定がサポートされています。デプロイ詳細にデフォルトのユーザー名やパスワードは表示されません。Broker のデータ統合を作成する前に、対象データベースに書き込み可能なユーザーを作成してください。

注意事項

古い EMQX Tables デプロイではカスタムユーザーや権限設定がサポートされていません。古いデプロイの場合は、デプロイ詳細に表示されているデフォルトの認証情報を使用してください。

EMQX Tables デプロイ内で、左メニューの User Management をクリックします。
+ Add User をクリックします。
ユーザー名とパスワードを入力します。パスワードは後から取得できずリセットのみ可能なので、必ず控えてください。
Privileges で Broker Integration を選択します。
このプリセットは EMQX ブローカーからのデータ取り込みに必要な権限を付与します。
Access Control で、このユーザーがアクセス可能なデータベースを選択します。クイックスタートでは public データベースを選択してください。カスタムデータベースを作成している場合はそちらを選択します。
Confirm をクリックします。

emqx_tables_add_user

作成したユーザーは User Management ページから管理可能です。

emqx_tables_user_entry

権限やアクセス制御モードの詳細は User Management を参照してください。

ブローカーと Tables 間の接続

接続方法はブローカーのデプロイタイプとネットワーク構成によって異なります：

デプロイタイプ	条件	接続方法	必要な操作	備考
Dedicated / Dedicated Flex	Tables と同じクラウドプラットフォーム、リージョン、ネットワーク	プライベート（セキュアで低レイテンシ）	2つ目のデプロイ作成時に Network Association で既存ネットワークを選択。プロジェクトのネットワーク管理ページで共有ネットワークを確認可能。	ネットワークには各サービスタイプ（Broker、Tables、Fleets、Agents）のデプロイが1つずつのみ配置可能。
Dedicated / Dedicated Flex	Tables と異なるリージョンまたはネットワーク	TLS を用いたパブリックインターネット	Broker デプロイの Network Management 設定で NAT Gateway を有効化	—
Serverless	—	TLS を用いたパブリックインターネット	操作不要	Network Association と NAT Gateway は適用されません。

データ取り込み用ルールの作成

ルールエンジンを使って MQTT メッセージを EMQX Tables に取り込み、永続化します。

Dedicated Flex（または Serverless）デプロイに入り、Data Integration に移動します。
初めてコネクターを作成する場合は、コネクター一覧から EMQX Tables を選択します。既にコネクターがある場合は + New Connector をクリックしてから EMQX Tables を選択します。
Quick Setup（デフォルト）を選択し、プロジェクト内の Tables デプロイを選択します。
接続設定を入力します：
- Database Name：public を使用
- Username：データ取り込み用ユーザーの作成で作成したユーザーを選択
- Password：該当ユーザーのパスワードを入力
Test をクリックして接続確認を行います。成功メッセージが表示されます。
New をクリックして New Rule を選択し、このコネクターを使ったルール作成に進みます。
SQL Editor で SQL ルールを定義します。
ルール SQL 例：
sql
```
SELECT
  timestamp as ts,
  payload.machine_id as machine_id,
  payload.production_line as production_line,
  payload.temperature as temperature,
  payload.vibration as vibration,
  payload.machine_status as machine_status
FROM "factory/+/metrics"
```
このルールは、トピック factory/+/metrics にマッチするすべての MQTT メッセージのペイロードからフィールドを抽出します。抽出した値にはエイリアスを付けて、ルールアクションの書き込み構文で参照可能にします。timestamp フィールドは ts にマッピングされ、EMQX Tables での時系列インデックスになります。
Next をクリックしてルールにアクションを追加します。アクション設定を行います：
- Connector：作成した EMQX Tables コネクターを選択
- Time Precision：millisecond
- Write Syntax：
  text
```
machine_metrics,production_line=${production_line},machine_id=${machine_id} temperature=${temperature},vibration=${vibration},machine_status=${machine_status} ${ts}
```
  この構文は、machine_metrics テーブルが存在しない場合は自動作成し、InfluxDB Line Protocol 形式でデータを書き込みます。
  - タグ：production_line、machine_id（ディメンションおよび主キーとして使用）
  - フィールド：temperature、vibration、machine_status（実際のメトリクス値）
  - タイムスタンプ：${ts} はメッセージから抽出した timestamp を使い時系列整合性を確保
Confirm をクリックしてルールを保存します。
Data Integration ページに戻ると、作成したコネクター、ルール、アクションが表示されます。

MQTT メッセージのパブリッシュ

クイックテストには、Dedicated Flex（または Serverless）デプロイ内蔵の診断ツールを使い、左メニューの Online Test をクリックします。

Online Test で、ユーザー名とパスワード、または自動生成された認証情報を使ってデプロイに接続します。
Messages セクションで以下の2つのメッセージを送信します：

TIP

テーブルが事前に作成されていない場合、EMQX Cloud は最初に正常に書き込まれたメッセージのデータ型をもとに自動でテーブルを作成します。テーブルとカラムが作成された後は、すべての書き込みが同じデータ型を使用する必要があり、異なる場合は書き込みに失敗します。

トピック：factory/A/metrics

ペイロード：

json

{
  "machine_id": "M001",
  "production_line": "A",
  "temperature": 36.5,
  "vibration": 0.03,
  "machine_status": "running"
}

json

{
  "machine_id": "M002",
  "production_line": "A",
  "temperature": 39.1,
  "vibration": 0.06,
  "machine_status": "warning"
}

publish_messages

EMQX Tables でのデータクエリ

EMQX Tables デプロイにアクセスします。
左メニューの Data Explorer をクリックします。
エディターに以下の SQL を入力し、Run Query をクリックします：
sql
```
SELECT * FROM machine_metrics;
```
取り込まれたメッセージが表示されます。

完成：ネイティブ MQTT から DB へのパイプライン

これでライブパイプラインが完成しました：

MQTT クライアント -> EMQX ブローカー -> ルールエンジン -> EMQX Tables -> SQL 分析

サードパーティのインフラ不要で、完全マネージドかつ時系列 IoT ワークロード向けに設計されています。

次は Grafana や Streamlit でメトリクスを可視化できます。詳細は Integration Guide をご覧ください。

データベース機能のクイックガイド

EMQX Broker からデフォルトの public データベース以外にも、カスタムデータベースやテーブルを定義し、SQL で手動挿入やクエリが可能です。テストや開発に柔軟に対応できます。

カスタムデータベースの作成

デフォルトの public データベースとは別にカスタムデータベースを作成できます。

デプロイ内の Data Explorer ページに移動します。
以下の SQL を入力し、Run Query をクリックします：
sql
```
CREATE DATABASE factory WITH (ttl='7d');
```

これにより、データ保持期間（TTL）が7日の factory という名前の新しいデータベースが作成されます。

create_custom_database

テーブルの作成

新しいデータベース内に工場メトリクス用の時系列テーブルを定義します。

Data Explorer で以下の SQL を入力し、Run Query をクリックします：

sql

CREATE TABLE factory.machine_metrics (
    ts TIMESTAMP NOT NULL,
    production_line STRING,
    machine_id STRING,
    temperature DOUBLE,
    vibration DOUBLE,
    machine_status STRING DEFAULT 'running',
    TIME INDEX (ts),
    PRIMARY KEY (production_line, machine_id)
) WITH (
    ttl='7d'
);

このテーブルは ts を時系列インデックスに、production_line と machine_id の複合主キーを持ちます。

SQL を使ったデータ挿入

EMQX Tables は SQL ベースと Line Protocol ベースの両方のデータ取り込みをサポートし、.txt や .lp ファイルのアップロードも可能です。

Data Explorer で以下のコマンドを実行し、サンプルデータを挿入します：

sql

INSERT INTO factory.machine_metrics (ts, production_line, machine_id, temperature, vibration, machine_status)
VALUES
    (now(), 'A', 'M001', 36.5, 0.03, 'running'),
    (now(), 'A', 'M002', 39.1, 0.06, 'warning'),
    (now(), 'B', 'M010', 37.2, 0.02, 'running'),
    (now(), 'B', 'M011', 45.6, 0.12, 'error');

now() は現在のタイムスタンプを挿入します。

データのクエリ

データ確認には Data Explorer を使用します。

クエリ例

すべてのレコードを表示：

sql

SELECT * FROM factory.machine_metrics;

query_all_records

直近60分の平均温度をラインと状態別に集計：

sql

SELECT production_line, machine_status, AVG(temperature) AS avg_temp
FROM factory.machine_metrics
WHERE ts > now() - INTERVAL '60 minute'
GROUP BY production_line, machine_status;

query_by_time_limits

デバイス別にフィルター：

sql

SELECT ts, temperature
FROM machine_metrics
WHERE machine_id = 'M001'
ORDER BY ts DESC
LIMIT 10;

query_filter_by_device

クイッククエリで高速アクセス

テーブルスキーマから SQL スニペットを素早く生成可能です：

Data Explorer の左側スキーマパネルで列にカーソルを合わせます。
列の横にある縦の省略記号（︙）アイコンをクリックします。
Quick Query を選択し、Query column、Query max、Query min などのオプションを使います。
生成された SQL が自動的にエディターに表示されます。

quick_query

リソース

サポートされている SQL ステートメントと句は Greptime ドキュメントをご覧ください。

より詳細なクエリ方法は EMQX Tables でのデータクエリを参照してください。

専用フレックス

VPC ピアリング接続

PrivateLink の設定

拡張認証

拡張認可

ルール

LLM ベースの MQTT データ処理

サーバレスデプロイメント API

Quick Start: Ingest MQTT Data into EMQX Tables ​

EMQX Tables 無料トライアル ​

クォータと期間 ​

有効期限 ​

ユースケース：スマートファクトリーモニタリング ​

EMQX ブローカーと EMQX Tables のデプロイ作成 ​

データ取り込み用ユーザーの作成 ​

ブローカーと Tables 間の接続 ​

データ取り込み用ルールの作成 ​

MQTT メッセージのパブリッシュ ​

EMQX Tables でのデータクエリ ​

完成：ネイティブ MQTT から DB へのパイプライン ​

データベース機能のクイックガイド ​

カスタムデータベースの作成 ​

テーブルの作成 ​

SQL を使ったデータ挿入 ​

データのクエリ ​

クエリ例 ​

クイッククエリで高速アクセス ​

リソース ​

Quick Start: Ingest MQTT Data into EMQX Tables

EMQX Tables 無料トライアル

クォータと期間

有効期限

ユースケース：スマートファクトリーモニタリング

EMQX ブローカーと EMQX Tables のデプロイ作成

データ取り込み用ユーザーの作成

ブローカーと Tables 間の接続

データ取り込み用ルールの作成

MQTT メッセージのパブリッシュ

EMQX Tables でのデータクエリ

完成：ネイティブ MQTT から DB へのパイプライン

データベース機能のクイックガイド

カスタムデータベースの作成

テーブルの作成

SQL を使ったデータ挿入

データのクエリ

クエリ例

クイッククエリで高速アクセス

リソース