PostgreSQLへのMQTTデータ取り込み

PostgreSQLは、世界で最も先進的なオープンソースのリレーショナルデータベースであり、シンプルなアプリケーションから複雑なデータ処理まで対応可能な強力なデータ処理能力を備えています。EMQXはPostgreSQLとの統合をサポートしており、IoTデバイスからのリアルタイムデータストリームを効率的に処理できます。この統合により、大規模なデータ保存、正確なクエリ、複雑なデータ関連分析が可能となり、データの整合性も確保されます。EMQXの効率的なメッセージルーティングとPostgreSQLの柔軟なデータモデルを活用することで、デバイスの状態監視、イベント追跡、操作監査が容易になり、ビジネスに深いデータインサイトと強力なビジネスインテリジェンス支援を提供します。

本ページでは、EMQXとPostgreSQL間のデータ統合について包括的に紹介し、ルールとシンクの作成方法を実践的に解説します。

TIP

本ページの内容はMatrixDBにも適用可能です。

動作概要

PostgreSQLデータ統合は、MQTTベースのIoTデータとPostgreSQLの強力なデータ保存機能を橋渡しするためにEMQXに標準搭載された機能です。組み込みのルールエンジンコンポーネントにより、EMQXからPostgreSQLへのデータ取り込みが簡素化され、複雑なコーディングを不要にします。

以下の図は、EMQXとPostgreSQL間のデータ統合の典型的なアーキテクチャを示しています。

EMQX Integration PostgreSQL

PostgreSQLへのMQTTデータ取り込みの流れは以下の通りです。

IoTデバイスがEMQXに接続：IoTデバイスがMQTTプロトコルを通じて正常に接続されると、オンラインイベントがトリガーされます。イベントにはデバイスID、送信元IPアドレスなどの情報が含まれます。
メッセージのパブリッシュと受信：デバイスは特定のトピックにテレメトリやステータスデータをパブリッシュします。EMQXはこれらのメッセージを受信すると、ルールエンジン内でマッチング処理を開始します。
ルールエンジンによるメッセージ処理：組み込みのルールエンジンは、特定のソースからのメッセージやイベントをトピックマッチングに基づいて処理します。ルールエンジンは対応するルールをマッチさせ、データ形式の変換、特定情報のフィルタリング、コンテキスト情報の付加などの処理を行います。
PostgreSQLへの書き込み：ルールがメッセージのPostgreSQLへの書き込みをトリガーします。SQLテンプレートを利用して、ルール処理結果からデータを抽出しSQLを構築、PostgreSQLに送信して実行することで、メッセージの特定フィールドを対応するテーブルやカラムに書き込んだり更新したりします。

イベントやメッセージデータがPostgreSQLに書き込まれた後は、PostgreSQLに接続してデータを読み出し、以下のような柔軟なアプリケーション開発が可能です。

Grafanaなどの可視化ツールに接続し、データに基づくグラフを生成してデータ変化を表示。
デバイス管理システムに接続し、デバイス一覧や状態を確認、異常動作を検知して潜在的な問題をタイムリーに解消。

特長とメリット

PostgreSQLは豊富な機能を持つ人気のオープンソースリレーショナルデータベースです。PostgreSQLとのデータ統合は、以下のような特長と利点をビジネスにもたらします。

柔軟なイベント処理：EMQXルールエンジンを通じて、PostgreSQLはデバイスのライフサイクルイベントを処理でき、IoTアプリケーション実装に必要な各種管理・監視タスクの開発を大幅に容易にします。イベントデータを分析することで、デバイスの故障や異常動作、傾向変化を迅速に検出し、適切な対応が可能です。
メッセージ変換：メッセージはEMQXルールで多様な処理・変換を経てからPostgreSQLに書き込まれるため、保存や利用がより便利になります。
柔軟なデータ操作：PostgreSQLデータブリッジが提供するSQLテンプレートを使い、特定フィールドのデータを対応テーブル・カラムに簡単に書き込み・更新でき、柔軟なデータ保存・管理を実現します。
ビジネスプロセスの統合：PostgreSQLデータブリッジにより、デバイスデータをPostgreSQLの豊富なエコシステムアプリケーションと統合可能で、ERP、CRM、その他カスタムビジネスシステムとの連携を促進し、高度なビジネスプロセスや自動化を実現します。
IoTとGIS技術の融合：PostgreSQLはGISデータの保存・クエリ機能を備え、地理空間インデックス、ジオフェンスやアラート、リアルタイム位置追跡、地理情報処理などをサポートします。EMQXの信頼性の高いメッセージ伝送能力と組み合わせることで、車両など移動体の地理位置情報を効率的に処理・分析し、リアルタイム監視、インテリジェントな意思決定、業務最適化を可能にします。
ランタイムメトリクス：各シンクのランタイムメトリクス（総メッセージ数、成功/失敗数、現在のレートなど）を閲覧可能です。

柔軟なイベント処理、多様なメッセージ変換、柔軟なデータ操作、リアルタイム監視・分析機能を通じて、効率的で信頼性が高くスケーラブルなIoTアプリケーションを構築でき、ビジネスの意思決定や最適化に役立ちます。

はじめる前に

このセクションでは、PostgreSQLデータベースシンクを作成する前に必要な準備として、PostgreSQLサーバーのセットアップやデータテーブルの作成方法について説明します。

前提条件

EMQXデータ統合のルールに関する知識
データ統合に関する知識

PostgreSQLのインストール

Dockerを使ってPostgreSQLをインストールし、Dockerイメージを起動します。

bash

# PostgreSQLのDockerイメージを起動し、パスワードをpublicに設定
docker run --name PostgreSQL -p 5432:5432 -e POSTGRES_PASSWORD=public -d postgres

# コンテナにアクセス
docker exec -it PostgreSQL bash

# コンテナ内でPostgreSQLサーバーに接続し、パスワードを入力
psql -U postgres -W

# データベースを作成し、選択
CREATE DATABASE emqx_data;

\c emqx_data;

データテーブルの作成

以下のSQL文を使い、PostgreSQLデータベースにクライアントID、トピック、ペイロード、メッセージの作成時間を保存するデータテーブルt_mqtt_msgを作成します。

sql

CREATE TABLE t_mqtt_msg (
  id SERIAL primary key,
  msgid character varying(64),
  sender character varying(64),
  topic character varying(255),
  qos integer,
  retain integer,
  payload text,
  arrived timestamp without time zone
);

クライアントID、イベントタイプ、作成時間を保存するデータテーブルemqx_client_eventsを作成するSQL文は以下の通りです。

sql

CREATE TABLE emqx_client_events (
  id SERIAL primary key,
  clientid VARCHAR(255),
  event VARCHAR(255),
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

コネクターの作成

PostgreSQLシンクを追加する前に、PostgreSQLコネクターを作成する必要があります。ここではEMQXとPostgreSQLがローカルマシンで動作していることを前提としています。リモートで動作している場合は設定を適宜調整してください。

EMQXダッシュボードにアクセスし、Integration -> Connectorをクリックします。
ページ右上のCreateをクリックします。
Create ConnectorページでPostgreSQLを選択し、Nextをクリックします。
コネクターの名前を入力します。名前は英数字の組み合わせとしてください（例：my_psql）。
接続情報を入力します。
- Server Host：127.0.0.1:5432（PostgreSQLサーバーがリモートの場合は実際のホスト名を入力）
- Database Name：emqx_data
- Username：postgres
- Password：public
- Enable TLS：暗号化接続を行う場合はトグルスイッチをオンにします。TLS接続の詳細は外部リソースアクセスのTLSを参照してください。
詳細設定（任意）：詳細はシンクの機能を参照してください。
Createをクリックする前に、Test ConnectivityをクリックしてコネクターがPostgreSQLサーバーに接続できるかテスト可能です。
ページ下部のCreateボタンをクリックしてコネクターの作成を完了します。ポップアップダイアログでBack to Connector Listをクリックしてコネクター一覧に戻るか、Create Ruleをクリックしてシンクを使ったルールの作成を続行できます。ルール作成の詳細はメッセージ保存用PostgreSQLシンク付きルールの作成およびイベント記録用PostgreSQLシンク付きルールの作成を参照してください。

注意事項

EMQX v5.7.1でDisable Prepared Statementsオプションが導入されました。PGBouncerのトランザクションモードやSupabaseなど、プリペアドステートメントをサポートしないPostgreSQLサービスを利用する場合は、詳細設定でこのオプションを有効にしてください。

メッセージ保存用PostgreSQLシンク付きルールの作成

このセクションでは、ダッシュボードでソースMQTTトピックt/#からのメッセージを処理し、処理済みデータを設定済みのシンク経由でPostgreSQLテーブルt_mqtt_msgに保存するルールの作成方法を示します。

ダッシュボードのIntegration -> Rulesページにアクセスします。
ページ右上のCreateをクリックします。
ルールIDにmy_ruleを入力し、SQLエディターにルールを入力します。ここではトピックt/#のMQTTメッセージをPostgreSQLに保存する例を示します。ルールのSELECT部分で選択するフィールドは、SQLテンプレートで使用する変数をすべて含むようにしてください。ルールSQLは以下の通りです。
sql
```
SELECT
*
FROM
"t/#"
```
TIP
初心者の方はSQL Examplesをクリックし、Enable Testを使ってSQLルールの学習とテストが可能です。
- Add Actionボタンをクリックし、ルールによりトリガーされるアクションを定義します。このアクションにより、EMQXはルールで処理したデータをPostgreSQLに送信します。
Type of ActionドロップダウンからPostgreSQLを選択し、ActionドロップダウンはデフォルトのCreate Actionのままにするか、既存のPostgreSQLアクションを選択します。本例では新規シンクを作成し、ルールに追加します。
シンクの名前と説明をフォームに入力します。
Connectorドロップダウンから先に作成したmy_psqlを選択します。ドロップダウン横のボタンから新規コネクター作成も可能です。設定パラメーターはコネクターの作成を参照してください。
SQL Templateを設定します。以下のSQL文を使ってデータを挿入します。
注意：これはプリプロセス済みSQLのため、フィールドは引用符で囲まず、文末にセミコロンを付けないでください。
sql
```
INSERT INTO t_mqtt_msg(msgid, sender, topic, qos, payload, arrived) VALUES(
  ${id},
  ${clientid},
  ${topic},
  ${qos},
  ${payload},
  TO_TIMESTAMP((${timestamp} :: bigint)/1000)
)
```
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。詳細はフォールバックアクションを参照してください。
詳細設定（任意）：詳細はシンクの機能を参照してください。
Createをクリックする前に、Test Connectivityを押してシンクがPostgreSQLサーバーに接続できるかテスト可能です。
Createボタンをクリックしてシンク設定を完了します。新しいシンクがAction Outputsに追加されます。
Create Ruleページに戻り、設定内容を確認後、Createボタンをクリックしてルールを生成します。

これでルールが正常に作成されました。Integration -> Rulesページで新規ルールを確認でき、**Action (Sink)**タブで新規PostgreSQLシンクも確認できます。

また、Integration -> Flow Designerを開くとトポロジーが表示され、トピックt/#のメッセージがルールmy_ruleで解析されPostgreSQLに書き込まれている様子を可視化できます。

イベント記録用PostgreSQLシンク付きルールの作成

このセクションでは、クライアントのオンライン/オフライン状態を記録し、イベントデータを設定済みシンク経由でPostgreSQLテーブルemqx_client_eventsに保存するルールの作成方法を示します。

手順はメッセージ保存用PostgreSQLシンク付きルールの作成とほぼ同様ですが、SQLテンプレートとSQLルールが異なります。

オンライン/オフライン状態記録用のルールSQL文は以下の通りです。

sql

SELECT
  *
FROM
  "$events/client_connected", "$events/client_disconnected"

イベント記録用のSQLテンプレートは以下の通りです。

注意：これはプリプロセス済みSQLのため、フィールドは引用符で囲まず、文末にセミコロンを付けないでください。

sql

INSERT INTO emqx_client_events(clientid, event, created_at) VALUES (
  ${clientid},
  ${event},
  TO_TIMESTAMP((${timestamp} :: bigint)/1000)
)

ルールのテスト

MQTTXを使ってトピックt/1にメッセージを送信し、オンライン/オフラインイベントをトリガーします。

bash

mqttx pub -i emqx_c -t t/1 -m '{ "msg": "hello PostgreSQL" }'

2つのシンクの稼働状況を確認します。メッセージ保存用シンクには新規の受信・送信メッセージが1件ずつあるはずです。イベント記録用シンクには2件のイベントレコードがあります。

t_mqtt_msgデータテーブルにデータが書き込まれているか確認します。

bash

emqx_data=# select * from t_mqtt_msg;
 id |              msgid               | sender | topic | qos | retain |            payload
        |       arrived
----+----------------------------------+--------+-------+-----+--------+-------------------------------+---------------------
  1 | 0005F298A0F0AEE2F443000012DC0002 | emqx_c | t/1   |   0 |        | { "msg": "hello PostgreSQL" } | 2023-01-19 07:10:32
(1 row)

emqx_client_eventsテーブルにデータが書き込まれているか確認します。

bash

emqx_data=# select * from emqx_client_events;
 id | clientid |        event        |     created_at
----+----------+---------------------+---------------------
  3 | emqx_c   | client.connected    | 2023-01-19 07:10:32
  4 | emqx_c   | client.disconnected | 2023-01-19 07:10:32
(2 rows)

Kubernetes

EMQX Operator

EMQX クラスター管理

API リファレンス

パスワード認証

監視

アクセスコントロール

統合

管理

Prometheus との統合

OpenTelemetry との統合

エンドツーエンドトレース

プラグインカタログ

JT/T 808 ゲートウェイ

GB/T 32960 ゲートウェイ

Volcengine RTC

シナリオ

GPT-Realtime

PostgreSQLへのMQTTデータ取り込み ​

動作概要 ​

特長とメリット ​

はじめる前に ​

前提条件 ​

PostgreSQLのインストール ​

データテーブルの作成 ​

コネクターの作成 ​

メッセージ保存用PostgreSQLシンク付きルールの作成 ​

イベント記録用PostgreSQLシンク付きルールの作成 ​

ルールのテスト ​

PostgreSQLへのMQTTデータ取り込み

動作概要

特長とメリット

はじめる前に

前提条件

PostgreSQLのインストール

データテーブルの作成

コネクターの作成

メッセージ保存用PostgreSQLシンク付きルールの作成

イベント記録用PostgreSQLシンク付きルールの作成

ルールのテスト