EMQX と ClickHouse の統合

EMQX は、高性能なリアルタイムメッセージ処理エンジンを備えたオープンソースの MQTT ブローカーであり、大規模な IoT デバイス向けのイベントストリーミングを支えています。高いスケーラビリティを持つ MQTT ブローカーとして、EMQX はあらゆるデバイスを、どのような規模でも接続するのに役立ちます。IoT データをどこへでも移動して処理できます。

EMQX Cloud は、EMQ が提供する IoT 分野向けの MQTT メッセージングミドルウェア製品です。世界初の完全マネージド型 MQTT 5.0 クラウドメッセージングサービスとして、EMQX Cloud は MQTT メッセージングサービス向けに、ワンストップの運用・保守と独自の分離環境を提供します。あらゆるモノがつながる時代において、EMQX Cloud は IoT 分野向けの業界アプリケーションを迅速に構築し、IoT データを容易に収集、伝送、コンピュート、永続化できるよう支援します。

クラウドプロバイダーが提供するインフラストラクチャ上で、EMQX Cloud は世界中の数十の国と地域にサービスを提供しており、5G および Internet of Everything アプリケーション向けに、低コストで安全かつ信頼性の高いクラウドサービスを提供しています。

• 非常に軽量な publish/subscribe 型メッセージ転送プロトコルとして設計された MQTT プロトコルに精通していること。
• 大規模な IoT デバイスのイベントストリーミングを支えるリアルタイムメッセージ処理エンジンとして、EMQX または EMQX Cloud を使用していること。
• デバイスデータを永続化するための ClickHouse Cloud インスタンスを準備していること。
• MQTT データをパブリッシュするために、EMQX Cloud のデプロイメントへ接続する MQTT クライアントのテストツールとして MQTT X を使用します。なお、MQTT ブローカーに接続できる他の方法でも問題ありません。

このセットアップでは、ClickHouse インスタンスを AWS の N. Virginia (us-east-1) にデプロイし、EMQX Cloud インスタンスも同じリージョンにデプロイしました。

セットアップ中は、接続設定にも注意する必要があります。このチュートリアルでは "Anywhere" を選択していますが、特定の場所に制限する場合は、EMQX Cloud のデプロイメントから取得した NAT ゲートウェイ の IP アドレスをホワイトリストに追加する必要があります。

次に、後で使用するためにユーザー名とパスワードを保存します。

その後、稼働中の ClickHouse インスタンスが利用可能になります。ClickHouse Cloud インスタンスの接続先アドレスを取得するには、"Connect" をクリックします。

EMQX Cloud とのインテグレーションのためのデータベースとテーブルを作成するには、"Connect to SQL Console" をクリックします。

以下の SQL ステートメントを参照するか、実際の状況に応じて SQL を調整してください。

CREATE TABLE emqx.temp_hum
(
   client_id String,
   timestamp DateTime,
   topic String,
   temp Float32,
   hum Float32
)
ENGINE = MergeTree()
PRIMARY KEY (client_id, timestamp)

EMQX Cloud で専用の MQTT ブローカーを作成するのは、数回クリックするだけです。

EMQX Cloud では、すべてのアカウントで、標準デプロイメントとプロフェッショナルデプロイメントの両方を 14 日間無料で試用できます。

EMQX Cloud を初めて利用する場合は、EMQX Cloud sign up ページにアクセスし、start free をクリックしてアカウントを登録します。

ログインしたら、アカウントメニューの「Cloud console」をクリックします。新しいデプロイメントを作成するための緑色のボタンが表示されます。

このチュートリアルでは、Professional デプロイメントを使用します。MQTT データをコードを1行も書かずに ClickHouse へ直接送信できるデータインテグレーション機能を提供しているのは、Pro バージョンだけだからです。

Pro バージョンを選択し、N.Virginial リージョンを選んで Create Now をクリックします。数分で、完全マネージド型の MQTT ブローカーが利用可能になります。

次に、パネルをクリックしてクラスター画面に移動します。このダッシュボードでは、MQTT ブローカーの概要を確認できます。

EMQX Cloud ではデフォルトで匿名接続が許可されていないため、MQTT クライアントツールでこのブローカーにデータを送信できるように、クライアント認証情報を追加する必要があります。

左側のメニューで 'Authentication & ACL' をクリックし、サブメニューで 'Authentication' をクリックします。右側の 'Add' ボタンをクリックして、後で MQTT 接続に使用するユーザー名とパスワードを設定します。ここでは、ユーザー名とパスワードに emqx と xxxxxx を使用します。

'Confirm' をクリックすると、完全マネージド型の MQTT ブローカーを使用する準備は完了です。

ClickHouse 連携の設定を開始する前に、まず NAT ゲートウェイを有効にする必要があります。デフォルトでは、MQTT ブローカーはプライベート VPC にデプロイされているため、パブリックネットワーク経由でサードパーティのシステムにデータを送信できません。

概要ページに戻り、ページの最下部までスクロールすると、NAT ゲートウェイのウィジェットが表示されます。Subscribe ボタンをクリックし、案内に従ってください。NAT ゲートウェイ は付加価値サービスですが、14 日間の無料トライアルも利用できます。

作成が完了すると、ウィジェット内にパブリック IP アドレスが表示されます。ClickHouse Cloud の設定時に Connect from a specific location を選択した場合は、この IP アドレスをホワイトリストに追加する必要がある点に注意してください。

EMQX Cloud Data Integrations は、EMQX のメッセージフローやデバイスイベントを処理し、それらに応答するためのルールを設定する際に使用されます。Data Integrations は、明確で柔軟な「設定可能な」アーキテクチャを提供するだけでなく、開発プロセスを簡素化し、ユーザビリティを向上させ、業務システムと EMQX Cloud の結合度を低減します。さらに、EMQX Cloud 固有の機能をカスタマイズするための優れたインフラストラクチャも提供します。

EMQX Cloud は、一般的なデータシステム向けに 30 を超えるネイティブインテグレーションを提供しています。ClickHouse もその 1 つです。

左側のメニューで「Data Integrations」をクリックし、「View All Resources」をクリックします。Data Persistence セクションにある ClickHouse を選ぶか、検索して ClickHouse を探します。

ClickHouse のカードをクリックして、新しいリソースを作成します。

• Note: このリソース用のメモを追加します。
• Server address: ClickHouse Cloud サービスのアドレスです。ポート番号も忘れずに含めてください。
• Database name: 上の手順で作成した emqx です。
• User: ClickHouse Cloud サービスへの接続に使用するユーザー名です。
• Key: 接続用のパスワードです。

リソースの作成中にポップアップが表示され、'New' をクリックすると、ルール作成ページに移動します。

EMQX には強力な ルールエンジン が用意されており、生の MQTT メッセージをサードパーティシステムに送信する前に、変換やデータの拡充を行えます。

このチュートリアルで使用するルールは次のとおりです。

SELECT
   clientid AS client_id,
   (timestamp div 1000) AS timestamp,
   topic AS topic,
   payload.temp AS temp,
   payload.hum AS hum
FROM
"temp_hum/emqx"

temp_hum/emqx トピックからメッセージを読み取り、client_id、topic、timestamp の情報を追加して JSONオブジェクトを拡充します。

したがって、トピックに送信する生の JSON は次のとおりです。

{"temp": 28.5, "hum": 0.68}

SQL テストを使用して、結果を確認できます。

次に、"NEXT" ボタンをクリックします。このステップでは、加工したデータを ClickHouse データベースにどのように挿入するかを EMQX Cloud に指定します。

リソースが 1 つしかない場合は、'Resource' と 'Action Type' を変更する必要はありません。 SQL テンプレートを設定するだけです。このチュートリアルで使用する例を以下に示します。

INSERT INTO temp_hum (client_id, timestamp, topic, temp, hum) VALUES ('${client_id}', ${timestamp}, '${topic}', ${temp}, ${hum})

これは ClickHouse にデータを挿入するためのテンプレートで、ここでは変数が使われています。

「Confirm」と「View Details」をクリックします。これで設定は完了です。ルールの詳細ページで、Data Integrationが正しく動作していることを確認できます。

temp_hum/emqx トピックに送信されたすべての MQTT メッセージは、ClickHouse Cloud のデータベースに永続化されます。

温度と湿度のデータをシミュレートし、まず MQTT X 経由で EMQX Cloud に送信します。その後、EMQX Cloud の Data Integrations を使って、そのデータを ClickHouse Cloud に保存します。

メッセージのパブリッシュには、任意の MQTT クライアントまたは SDK を使用できます。このチュートリアルでは、EMQ が提供する使いやすい MQTT クライアントアプリケーション MQTT X を使用します。

MQTTX で "New Connection" をクリックし、接続フォームに入力します。

• Name: 接続名。任意の名前を指定します。
• Host: MQTT ブローカーの接続先アドレス。EMQX Cloud の概要ページで確認できます。
• Port: MQTT ブローカーの接続ポート。EMQX Cloud の概要ページで確認できます。
• Username/Password: 上で作成した認証情報を使用します。このチュートリアルでは emqx と xxxxxx です。

右上の "Connect" ボタンをクリックすると、接続が確立されます。

これで、このツールを使って MQTT ブローカーにメッセージを送信できます。 入力内容:

1. ペイロードのフォーマットを "JSON" に設定します。
2. トピックを temp_hum/emqx に設定します (先ほどルールで設定したトピックです)
3. JSON 本文:

{"temp": 23.1, "hum": 0.68}

右側の送信ボタンをクリックします。温度の値を変更して、MQTT ブローカーにさらにデータを送信できます。

EMQX Cloud に送信されたデータは、ルールエンジンで処理され、ClickHouse Cloud に自動的に挿入されます。

ルール監視を確認し、成功数が1増えていることを確認します。

それでは、ClickHouse Cloud 上のデータを確認してみましょう。理想的には、MQTTX で送信したデータは EMQX Cloud に送られ、native データインテグレーションによって ClickHouse Cloud のdatabaseに永続化されます。

ClickHouse Cloud のパネルにある SQL Console に接続するか、任意のクライアントツールを使用して ClickHouse からデータを取得できます。このチュートリアルでは、SQL Console を使用します。 次の SQL を実行します。

SELECT * FROM emqx.temp_hum;

コードを一切書かずに、MQTT データを EMQX Cloud から ClickHouse Cloud へ転送できるようになりました。EMQX Cloud と ClickHouse Cloud を使えば、インフラを管理する必要はなく、データを ClickHouse Cloud に安全に保存しながら、IoT アプリケーションの開発に集中できます。