EMQX Cloud

日本語

简体中文
English

日本語

简体中文
English

Appearance

お問い合わせ
お問い合わせ

クイックスタート：MQTTデータを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を例にしていますが、サーバレスの場合もデータ統合の手順は同じです。ただし、サーバレスはTLSを用いてパブリックインターネット経由でEMQX Tablesに接続するため、Network AssociationやNAT Gatewayは適用されません。

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

  1. EMQX Cloudコンソールにログインします。

  2. 既存のプロジェクトを選択するか、新規作成します。

  3. EMQX Brokersの下で、+ New Deploymentをクリックします。

  4. Dedicated Flexプランを選択し、仕様を設定します。

    • 必要に応じてCloud Provider & Regionを選択します。
    • クイックデモのため、以下のオプションはデフォルトのままにします。
      • Tier
      • Smart Data Hubのサブスクライブ（任意）
      • デプロイメント名とプロジェクト
      • EMQXバージョン

    右下のDeployをクリックします。

    create_broker_deployment

    詳細はDedicated Flexデプロイの作成を参照してください。

  5. EMQX Tablesの下で、+ New Deploymentをクリックします。

  6. デフォルトのStarterプランを選択し、ブローカーと同じクラウドプロバイダーとリージョンを選択します。

  7. （任意）Network Associationで既存のネットワークをドロップダウンから選択します。同じクラウドプラットフォームかつ同リージョンのため、ブローカーのネットワークがリストに表示されます。選択すると両デプロイメントが同じネットワークを共有し、プライベート接続で通信可能になります。

  8. Tierを選択します。

  9. （任意）Deployment Nameを入力します。

  10. Deployをクリックします。詳細はEMQX Tablesデプロイの作成を参照してください。

new_table_deployment

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

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

データ取り込み用ユーザーを作成する ​

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

注意

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

  1. EMQX Tablesデプロイメントの左メニューからUser Managementをクリックします。

  2. + Add Userをクリックします。

  3. ユーザー名とパスワードを入力します。パスワードは後から取得できずリセットのみ可能なので、必ず控えてください。

  4. PrivilegesでBroker Integrationを選択します。

    このプリセットはEMQX ブローカーからのデータ取り込みに必要な権限を付与します。

  5. Access Controlでこのユーザーがアクセス可能なデータベースを選択します。クイックスタートではpublicデータベースを選択してください。カスタムデータベースを作成した場合はそちらを選択します。

  6. Confirmをクリックします。

emqx_tables_add_user

作成したユーザーはUser Managementページで管理できます。

emqx_tables_user_entry

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

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

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

デプロイタイプ条件接続方法必要な操作備考
Dedicated / Dedicated FlexTablesと同じクラウドプラットフォーム、リージョン、ネットワークプライベート（安全で低レイテンシ）2つ目のデプロイ作成時にNetwork Associationで既存ネットワークを選択。プロジェクトのネットワーク管理ページで共有ネットワークを確認可能。1つのネットワークには最大1つのブローカーと1つのTablesデプロイが対応可能。
Dedicated / Dedicated FlexTablesと異なるリージョンまたはネットワークTLSを用いたパブリックインターネット経由ブローカーのNetwork Management設定でNAT Gatewayを有効化—
Serverless—TLSを用いたパブリックインターネット経由操作不要Network AssociationとNAT Gatewayは非適用。

データ取り込み用ルールを作成する ​

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

  1. Dedicated Flex（またはServerless）デプロイメントにアクセスし、Data Integrationに移動します。

  2. 初めてのコネクターの場合は、コネクター一覧からEMQX Tablesを選択します。既にコネクターがある場合は**+ New Connector**をクリックし、EMQX Tablesを選択します。

    create_connector

  3. Quick Setup（デフォルト）を選択し、プロジェクト内のTablesデプロイメントを選択します。

  4. 接続設定を入力します：

    connector_quick_setup

  5. Testをクリックして接続を確認します。成功メッセージが表示されます。

  6. Newをクリックし、New Ruleを選択してこのコネクターを使ったルール作成に進みます。

  7. 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にレコードを保存する際の時間インデックスになります。

  8. 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を使い時系列を整合

  9. Confirmをクリックしてルールを保存します。

    Data Integrationページに戻ると、作成したコネクター、ルール、アクションが表示されます。

    create_rule

MQTTメッセージをパブリッシュする ​

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

  1. Online Testで、ユーザー名とパスワードまたは自動生成された認証情報でデプロイメントに接続します。

  2. 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でデータをクエリする ​

  1. EMQX Tablesデプロイメントにアクセスします。

  2. 左メニューのData Explorerをクリックします。

  3. エディタに以下のSQLを入力し、Run Queryをクリックします：

    sql
    SELECT * FROM machine_metrics;

    取り込んだメッセージが表示されます。

    query_test_data

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

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

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

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

次はGrafanaやStreamlitでメトリクスを可視化できます。詳細は統合ガイドをご覧ください。

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

EMQX ブローカーからデフォルトのpublicデータベースにデータを取り込むだけでなく、EMQX Tablesではカスタムデータベースやテーブルの定義、SQLによる手動データ挿入、内蔵のData Explorerインターフェースでのクエリも可能です。これによりテストや開発の柔軟性が向上します。

カスタムデータベースを作成する ​

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

  1. デプロイメントのData Explorerページに移動します。

  2. 以下の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スニペットを素早く生成できます：

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

quick_query

リソース ​

GreptimeのサポートされているSQL文と句を参照してください。

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