AlloyDB に MQTT データを取り込む
AlloyDB for PostgreSQL は、Google Cloud が提供する完全マネージドの PostgreSQL 互換データベースサービスで、要求の厳しいエンタープライズワークロード向けに設計されています。EMQX は AlloyDB とのシームレスな統合をサポートしており、IoT デバイスからの MQTT データをリアルタイムで取り込み、保存することが可能です。EMQX の効率的なメッセージルーティングと AlloyDB の高スループットなトランザクション処理能力、および HTAP（ハイブリッドトランザクショナル／アナリティカルプロセッシング）エンジンによるリアルタイム分析を活用することで、デバイスの状態取得、イベントログ記録、洞察に富んだ分析を実現する強力なパイプラインを構築できます。
本ページでは、EMQX と AlloyDB 間のデータ統合について包括的に紹介し、データ統合の作成および検証に関する実践的な手順を提供します。
動作概要
EMQX における AlloyDB データ統合は組み込み機能であり、MQTT ベースの IoT データストリームを AlloyDB の高性能な PostgreSQL 互換データベースに直接取り込みます。組み込みのルールエンジンコンポーネントを利用することで、EMQX から AlloyDB へのデータ取り込みと分析が簡素化され、複雑なコーディングを不要にします。AlloyDB シンクを通じて、MQTT メッセージやクライアントイベントを AlloyDB に保存可能です。また、イベントにより AlloyDB 内のデータの更新や削除操作をトリガーでき、デバイスのオンライン状態や接続履歴などの情報を記録できます。
以下の図は、EMQX と AlloyDB 間のデータ統合の典型的なアーキテクチャを示しています。
AlloyDB への MQTT データ取り込みは以下のように動作します。
- IoT デバイスが EMQX に接続：IoT デバイスが MQTT プロトコルを介して正常に接続されると、オンラインイベントがトリガーされます。イベントにはデバイスID、送信元IPアドレス、その他属性情報が含まれます。
- メッセージのパブリッシュと受信：デバイスは特定のトピックにテレメトリや状態データをパブリッシュします。EMQX がこれらのメッセージを受信すると、ルールエンジン内でマッチング処理が開始されます。
- ルールエンジンによるメッセージ処理：EMQX のルールエンジンは、トピックやメッセージ内容に基づいて定義されたルールとイベントやメッセージをマッチングし処理します。処理内容には、データ変換（例：JSON から SQL 用フォーマットへの変換）、フィルタリング、コンテキスト情報によるデータ強化などが含まれ、データベース挿入前に行われます。
- AlloyDB への書き込み：マッチしたルールにより AlloyDB に対して SQL が実行されます。SQL テンプレートを使用して、処理済みデータのフィールドを AlloyDB のテーブルやカラムにマッピングできます。AlloyDB は並列クエリ実行と組み込みのカラムナーエンジンによる最適化ストレージをサポートしているため、高速にデータを挿入しつつ即座に分析可能です。
イベントおよびメッセージデータが AlloyDB に書き込まれた後は、AlloyDB に接続して柔軟なアプリケーション開発が可能です。例として：
- Grafana などの可視化ツールに接続し、データに基づくチャートを生成してデータ変化を表示
- AlloyDB とデバイス管理システムや分析モデルを統合し、デバイスの健全性監視、異常検知、アラート発動
- AlloyDB の HTAP 機能を活用し、ライブの IoT データに対して集計、結合、時系列クエリなどの複雑な分析を実行しつつ、新しいデバイステレメトリをリアルタイムで処理
特長とメリット
AlloyDB とのデータ統合により、以下の特長と利点をビジネスにもたらします。
- 柔軟なイベント処理：EMQX のルールエンジンを用いることで、デバイスのライフサイクルイベント（接続、切断、状態変化）を低レイテンシで AlloyDB に保存・処理可能です。AlloyDB の並列クエリ実行と独立スケーリングと組み合わせることで、リアルタイムにイベントデータを分析し、デバイス障害や異常、利用傾向を検出できます。
- メッセージ変換：EMQX ルールでメッセージを高度に処理・変換してから AlloyDB に書き込むため、保存や利用がより便利になります。
- SQL テンプレートによる柔軟なデータ操作：EMQX の SQL テンプレートマッピングを通じて、構造化された IoT データを AlloyDB のテーブル・カラムに挿入または更新できます。AlloyDB の PostgreSQL 互換性により標準 SQL、JSONB ストレージ、インデックスが利用可能で、AI によるインデックス自動最適化でクエリ性能がワークロードの変化に応じて向上します。
- 業務プロセス統合：AlloyDB の PostgreSQL エコシステムとの互換性により、Google Cloud 上またはオンプレミスの ERP、CRM、GIS、カスタム業務システムと直接統合可能です。EMQX と組み合わせることで、複雑なデータパイプラインなしにイベント駆動の自動化や業務プロセスオーケストレーションを実現できます。
- 高度な地理空間機能：PostGIS などの PostgreSQL 拡張機能を通じて、AlloyDB は地理空間データの保存、インデックス、クエリをサポートし、ジオフェンシング、ルート追跡、位置情報分析を可能にします。EMQX の信頼性の高い MQTT 取り込みと組み合わせることで、車両追跡、資産監視、リアルタイム IoT-GIS ソリューションの構築が可能です。
- 組み込みのメトリクスと監視：EMQX は各 AlloyDB シンクのランタイムメトリクスを提供し、AlloyDB は Cloud Monitoring と統合してクエリ性能、ストレージ利用率、レプリカの健全性を監視し、エンドツーエンドの可観測性を確保します。
はじめる前に
本セクションでは、AlloyDB 統合の作成を開始する前に必要な準備、すなわち AlloyDB インスタンスの作成やデータベースおよびデータテーブルの作成方法について説明します。
前提条件
AlloyDB でのデータベースとテーブルの作成
EMQX で AlloyDB コネクターを作成する前に、AlloyDB インスタンスが利用可能であり、IoT データを保存するためのデータベースおよびテーブルが作成されていることを確認してください。
公式 AlloyDB クイックスタートガイド に従って以下を実施します。
AlloyDB インスタンスを作成します。
このセットアップ時に、以下のようにデータベースユーザーの認証情報を定義します。
ユーザー名：
emqx_user（接続、挿入、更新、選択の権限を持つ必要があります）パスワード：
your_password_here
このユーザーはインスタンスのプロビジョニング時に作成するか、後から SQL、Google Cloud コンソール、または
gcloudCLI で作成可能です。
インスタンス内にデータベースを作成します。例としてデータベース名は
emqx_dataとします。上記の認証情報を使い、
psqlなどの PostgreSQL 互換クライアントでデータベースに接続します。MQTT メッセージとクライアントイベントデータを保存するために、
emqx_dataデータベース内に2つのテーブルを作成します。以下の SQL 文で、クライアントID、トピック、QoS、ペイロード、到着時間などのメタデータを含む MQTT メッセージを保存する
t_mqtt_msgテーブルを作成します。sqlCREATE 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 );以下の SQL 文で、クライアントのオンライン／オフラインイベントをタイムスタンプ付きで保存する
emqx_client_eventsテーブルを作成します。sqlCREATE TABLE emqx_client_events ( id SERIAL primary key, clientid VARCHAR(255), event VARCHAR(255), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );
AlloyDB コネクターの作成
AlloyDB シンクを追加する前に、EMQX で AlloyDB コネクターを作成します。コネクターは EMQX が Google Cloud 上の AlloyDB インスタンスに接続する方法を定義します。
EMQX ダッシュボードで、Integration -> Connector に移動します。
ページ右上の Create をクリックします。
Create Connector ページで AlloyDB を選択し、Next をクリックします。
コネクター名を入力します。名前は英数字で始まり、英数字、ハイフン、アンダースコアを含めることができます。例：
my_alloydb接続情報を入力します。
- Server Host：Google Cloud 上の AlloyDB インスタンスのホスト名または IP アドレス
- Database Name：EMQX がデータを書き込む AlloyDB の対象データベース名。例：
emqx_data - Username：認証および識別に使用する AlloyDB のデータベースユーザー名。例：
emqx_user - Password：
emqx_userのパスワード - Enable TLS：暗号化接続を確立する場合はトグルをオンにします。TLS 接続の詳細は TLS for External Resource Access を参照してください。
詳細設定（任意）：接続プールサイズ、アイドルタイムアウト、リクエストタイムアウトなどの追加接続プロパティを設定できます。
Test Connectivity をクリックして、EMQX が指定した設定で AlloyDB インスタンスに正常に接続できるか検証します。
Create をクリックしてコネクターを保存します。
作成後は以下のいずれかを選択できます。
- Back to Connector List をクリックして全コネクターを表示
- Create Rule をクリックして、このコネクターを使った AlloyDB へのデータ転送ルールをすぐに作成
詳細な例は以下を参照してください。
メッセージ保存用 AlloyDB シンクを使ったルールの作成
このセクションでは、ソース MQTT トピック t/# からのメッセージを処理し、処理済みデータを設定済み AlloyDB テーブル t_mqtt_msg に保存するルールをダッシュボードで作成する方法を示します。
ダッシュボードの Integration -> Rules ページに移動します。
ページ右上の Create をクリックします。
ルール ID に
my_ruleを入力し、SQL エディターにルールを記述します。ここではトピックt/#の MQTT メッセージを AlloyDB に保存するため、ルールの SELECT 部分で SQL テンプレートで使用するすべての変数を含むフィールドを選択してください。ルール SQL は以下の通りです。sqlSELECT * FROM "t/#"TIP
初心者の方は SQL Examples をクリックし、Enable Test を有効にして SQL ルールの学習とテストを行うことができます。
- Add Action ボタンをクリックし、ルールによってトリガーされるアクションを定義します。このアクションにより、EMQX はルールで処理したデータを AlloyDB に送信します。
Type of Action ドロップダウンから AlloyDB を選択し、Action ドロップダウンはデフォルトの
Create Actionのままにするか、既存の AlloyDB アクションを選択できます。本例では新規シンクを作成しルールに追加します。シンクの名前と説明をフォームに入力します。
Connector ドロップダウンから、前に作成した
my_alloydbを選択します。新規コネクターはドロップダウン横のボタンから作成可能です。設定パラメーターはAlloyDB コネクターの作成を参照してください。SQL Template を設定します。以下の SQL 文を使ってデータを挿入します。
注意：これはプリプロセス済み SQLですので、フィールドは引用符で囲まず、文末にセミコロンを付けないでください。
sqlINSERT 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 をクリックしてシンクが AlloyDB インスタンスに接続できるかテストできます。
Create ボタンをクリックしてシンク設定を完了します。新しいシンクが Action Outputs に追加されます。
Create Rule ページで設定内容を確認し、Save ボタンをクリックしてルールを生成します。
これでルールが正常に作成されました。Integration -> Rules ページで新規ルールを確認でき、Action (Sink) タブで新規 AlloyDB シンクも確認できます。
また、Integration -> Flow Designer でトポロジーを確認でき、トピック t/# のメッセージがルール my_rule によって解析され AlloyDB に書き込まれている様子を可視化できます。
イベント記録用 AlloyDB シンクを使ったルールの作成
このセクションでは、クライアントのオンライン／オフライン状態を記録し、イベントデータを設定済み AlloyDB テーブル emqx_client_events に保存するルールの作成方法を示します。
手順はメッセージ保存用 AlloyDB シンクを使ったルールの作成とほぼ同様で、SQL テンプレートと SQL ルールのみ異なります。
オンライン／オフライン状態記録用の SQL ルールは以下の通りです。
SELECT
*
FROM
"$events/client_connected", "$events/client_disconnected"イベント記録用の SQL テンプレートは以下の通りです。
注意：これはプリプロセス済み SQLですので、フィールドは引用符で囲まず、文末にセミコロンを付けないでください。
INSERT INTO emqx_client_events(clientid, event, created_at) VALUES (
${clientid},
${event},
TO_TIMESTAMP((${timestamp} :: bigint)/1000)
)ルールのテスト
MQTTX を使ってトピック t/1 にメッセージを送信し、オンライン／オフラインイベントをトリガーします。
mqttx pub -i emqx_c -t t/1 -m '{ "msg": "hello AlloyDB" }'2つのシンクの稼働状況を確認します。メッセージ保存用シンクでは新規の受信メッセージと送信メッセージがそれぞれ1件ずつあるはずです。イベント記録用シンクでは2件のイベントレコードが存在します。
t_mqtt_msg データテーブルにデータが書き込まれているか確認します。
emqx_data=# select * from t_mqtt_msg;
id | msgid | sender | topic | qos | retain | payload
| arrived
----+----------------------------------+--------+-------+-----+--------+-------------------------------+---------------------
1 | 0005F298A0F0AEE2F443000012DC0002 | emqx_c | t/1 | 0 | | { "msg": "hello AlloyDB" } | 2023-01-19 07:10:32
(1 row)emqx_client_events テーブルにデータが書き込まれているか確認します。
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)