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プランを選択し、仕様を設定します。

    • 必要に応じてクラウドプロバイダーとリージョンを選択します。
    • 以下のオプションはデフォルトのままで素早くデモを行えます。
      • 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で既存ネットワークを選択。プロジェクトのネットワーク管理で共有ネットワークを確認可能。ネットワークには各サービス種別（ブローカー、Tables、Fleets、Agents）ごとに1つのデプロイのみ配置可能。
Dedicated / Dedicated FlexTablesと異なるリージョンまたはネットワークTLSを用いたパブリックインターネットブローカーのネットワーク管理設定で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メッセージのペイロードからフィールドを抽出します。抽出した値にはエイリアスを付けて、ルールアクションのWrite Syntax設定で参照可能にしています。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は最初に正常に書き込まれたメッセージのデータ型に基づいて自動的にテーブルを作成します。テーブルとカラムが作成された後は、すべての書き込みで同じデータ型を使用する必要があり、異なる場合は書き込みが失敗します。

  • Topic：factory/A/metrics

  • Payload：

    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 Tablesでは、デフォルトのpublicデータベース以外にカスタムデータベースやテーブルを定義し、SQLで手動挿入やクエリを行うことも可能です。テストや開発に柔軟に対応できます。

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

デフォルトの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の複合主キーを持ちます。

データの挿入 ​

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

オプション1：SQLでデータ挿入 ​

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()は現在のタイムスタンプを挿入します。

オプション2：InfluxDB Line Protocolでデータ挿入 ​

  1. Import Dataに移動します。

  2. 以下のLine Protocolをエディターに貼り付けます：

    text
    machine_metrics,production_line=A,machine_id=M001 temperature=36.5,vibration=0.03,machine_status="running"
machine_metrics,production_line=A,machine_id=M002 temperature=39.1,vibration=0.06,machine_status="warning"
machine_metrics,production_line=B,machine_id=M010 temperature=37.2,vibration=0.02,machine_status="running"
machine_metrics,production_line=B,machine_id=M011 temperature=45.6,vibration=0.12,machine_status="error"

  3. タイムスタンプの精度（例：Nanoseconds）を選択します。

  4. Writeをクリックしてデータをインポートします。

この方法はタグを主キーに、フィールドをメトリクス値に自動マッピングします。

Line Protocolのマッピング：

  • Measurement：テーブル名
  • Tags：ディメンション（例：production_line、machine_id）
  • Fields：メトリクス（例：temperature、vibration、machine_status）
  • Timestamp：任意。指定がない場合はEMQX Tablesが現在時刻を使用。

オプション3：.txtまたは.lpファイルのアップロード ​

  1. Import Dataページのエディター上部のアップロードアイコンをクリックします。

  2. .lpテンプレートをダウンロードします。

  3. Line Protocolエントリを編集します。

  4. ファイル（.txtまたは.lp、最大10MB）をアップロードします。

  5. Importをクリックします。

データのクエリ ​

データを確認するには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

リソース ​

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

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