Confluent Cloud への MQTT データストリーム

Confluent Cloud は、Apache Kafka をベースにしたフルマネージドのイベントストリーミングプラットフォームであり、完全に管理されたサービスとして提供されます。本チュートリアルでは、EMQX Cloud で Confluent Cloud とのデータ統合を作成することで、MQTT データを Confluent Cloud にストリームする方法を紹介します。このデータ統合を通じて、クライアントは MQTT プロトコルを使用して温度および湿度データを EMQX Cloud に報告し、そのデータを Confluent Cloud にストリームできます。また、本チュートリアルでは、MQTTX を使用してデータ統合をテストする方法も示します。

本ページでは、Confluent データ統合の機能的特徴を詳細に紹介し、作成手順について実践的なガイダンスを提供します。内容は、Kafka コネクターの作成、ルールの作成、ルールのテストを含みます。シミュレートされた温度および湿度データを MQTT プロトコル経由で EMQX Cloud に報告し、設定されたデータ統合を通じて Confluent にデータを保存する方法を示します。

動作の仕組み

Confluent データ統合は、EMQX のすぐに使える機能であり、MQTT ベースの IoT データと Confluent の強力なデータ処理機能を橋渡しします。組み込みのルールエンジンコンポーネントを使用することで、両プラットフォーム間のデータフローと処理を簡素化し、複雑なコーディングを不要にします。

Confluent へのメッセージデータ転送の基本的なワークフローは以下の通りです：

1. メッセージのパブリッシュと受信：車両に接続された IoT デバイスは MQTT プロトコルを介して EMQX に正常に接続し、定期的に状態データを含むメッセージをパブリッシュします。EMQX がこれらのメッセージを受信すると、ルールエンジン内でマッチング処理を開始します。
2. メッセージデータの処理：これらの MQTT メッセージは、組み込みのルールエンジンとメッセージングサーバーの協調動作により、トピックマッチングルールに従って処理されます。メッセージが到着してルールエンジンを通過すると、事前に定義された処理ルールが評価されます。ペイロード変換を指定するルールがあれば、データ形式の変換、特定情報のフィルタリング、追加コンテキストによるペイロードの拡充などの変換が適用されます。
3. Confluent へのブリッジ：ルールエンジンで定義されたルールは、メッセージを Confluent に転送するアクションをトリガーします。Confluent 機能を使用して、MQTT トピックは Confluent の事前定義された Kafka トピックにマッピングされ、すべての処理済みメッセージとデータがこれらのトピックに書き込まれます。

特徴と利点

Confluent とのデータ統合は、以下の特徴と利点をビジネスにもたらします：

• 大規模メッセージ伝送の信頼性：EMQX と Confluent Cloud はいずれも高信頼のクラスター機構を使用し、安定かつ信頼性の高いメッセージ伝送チャネルを確立します。大規模な IoT デバイスからのメッセージ損失をゼロに保証します。両者ともノード追加による水平スケールが可能であり、リソースを動的に調整して突発的な大規模メッセージに対応し、メッセージ伝送の可用性を確保します。
• 強力なデータ処理能力：EMQX のローカルルールエンジンと Confluent Cloud はいずれも信頼性の高いストリーミングデータ処理能力を提供し、デバイスからアプリケーションまでの異なる段階で作用します。リアルタイムのデータフィルタリング、形式変換、集約分析などをシナリオに応じて実施し、より複雑な IoT メッセージ処理ワークフローを実現し、データ分析アプリケーションのニーズに応えます。
• 強力な統合機能：Confluent Cloud が提供する各種コネクターを通じて、EMQX は他のデータベース、データウェアハウス、データストリーム処理システムなどと容易に統合でき、完全な IoT データワークフローを構築し、俊敏なデータ分析アプリケーションを支援します。
• 高スループット処理能力：同期・非同期の両書き込みモードをサポートし、リアルタイム優先とパフォーマンス優先のデータ書き込み戦略を区別でき、異なるシナリオでレイテンシとスループットのバランスを柔軟に調整できます。
• 効果的なトピックマッピング：ブリッジ設定を通じて、多数の IoT ビジネストピックを Kafka トピックにマッピング可能です。EMQX は MQTT ユーザープロパティを Kafka ヘッダーにマッピングすることをサポートし、1対1、1対多、多対多の柔軟なトピックマッピング方法を採用し、MQTT トピックフィルター（ワイルドカード）もサポートします。

これらの特徴は統合能力と柔軟性を高め、効果的かつ堅牢な IoT プラットフォームアーキテクチャの構築を支援します。増大する IoT データは安定したネットワーク接続を介して伝送され、さらに効果的に保存・管理されます。

はじめる前に

このセクションでは、EMQX Cloud コンソールで Confluent データ統合を設定するための準備作業について説明します。

前提条件

• ルールについて理解していること。
• データ統合について理解していること。

Confluent Cloud クラスターのセットアップ

Confluent データ統合を作成する前に、Confluent Cloud コンソールで Confluent クラスターを作成し、Confluent Cloud CLI を使用してトピックと API キーを作成する必要があります。

クラスターには4種類があり、それぞれ異なるネットワークアクセスを持ちます。詳細は Confluent ドキュメント を参照してください。クラスターの種類により、デプロイに必要なネットワーク環境が異なります。

機能	Basic	Standard	Enterprise	Dedicated
パブリックネットワーク	YES	YES	NO	YES
プライベートネットワーク	NO	NO	YES	YES

以下のセクションでは、パブリックネットワークおよびプライベートネットワークでの Confluent Cloud クラスターのセットアップ方法を示します。

パブリックネットワークでのクラスターセットアップ

Basic / Standard / Dedicated クラスターをお持ちの場合、または作成したい場合は、パブリックネットワークの設定が可能です。

クラスターの作成

1. Confluent Cloud コンソールにログインし、クラスターを作成します。本デモでは Basic クラスターを例に選択し、Begin Configuration をクリックします。
2. リージョン／ゾーンを選択します。レイテンシを考慮し、EMQX Cloud のデプロイリージョンは Confluent Cloud のリージョンと一致させることを推奨します。Continue をクリックします。
3. クラスター名を入力し、Launch cluster をクリックします。これでクラウド上に稼働中のクラスターが作成されます。

API キーの作成

1. ナビゲーションメニューから API Keys をクリックし、Add key をクリックします。
2. API キーのスコープを選択します。Global access を選択し、Next をクリックします。
3. API キーを作成し、後の設定のためにキーをダウンロードします。

EMQX Cloud で NAT ゲートウェイを有効化

1. EMQX Cloud コンソールにログインし、デプロイの概要ページに入ります。
2. ページ下部の NAT Gateway タブをクリックし、Subscribe Now をクリックします。詳細は NAT Gateway を参照してください。

上記の手順で、パブリックネットワークの前提設定が完了します。

プライベートネットワークでのクラスターセットアップ

Enterprise / Dedicated クラスターをお持ちの場合、または作成したい場合は、プライベートネットワークの設定が可能です。Enterprise クラスターは Private Link のみ対応し、Dedicated クラスターは Private Link と VPC Peering の両方をサポートします。

1. Confluent Cloud コンソールにログインし、クラスターを作成します。本デモでは Dedicated クラスターを例に選択し、Begin Configuration をクリックします。

2. リージョン／ゾーンを選択します。EMQX Cloud のデプロイリージョンが Confluent Cloud のリージョンと一致していることを確認し、Continue をクリックします。

3. ネットワーク設定で VPC Peering を選択し、このクラスターが VPC ピアリング接続のみでアクセス可能になるようにします。

4. クラスター用の CIDR ブロックを指定し、Continue をクリックします。3つの可用性ゾーンそれぞれに CIDR 範囲を指定した場合は、3つすべての範囲を含むサポートチケットを提出する必要があります。SRE チームが接続に必要な追加設定を行います。

5. ニーズに応じて暗号化キーの管理方法を選択し、Continue をクリックします。

6. カードのバインド後、クラスターの起動準備が整います。

Confluent Cloud CLI によるクラスター管理

Confluent Cloud でクラスターが稼働したら、Confluent Cloud CLI を使って管理できます。以下は基本的なコマンド例です。

Confluent Cloud CLI のインストール

curl -sL --http1.1 https://cnfl.io/cli | sh -s -- -b /usr/local/bin

既に CLI をインストール済みの場合は、以下で更新可能です：

confluent update

アカウントにログイン

confluent login --save

環境を選択

confluent environment use env-xxxxx

クラスターを選択

confluent kafka cluster use lkc-xxxxx

API キーとシークレットの使用

既存の API キーを CLI に追加する場合：

confluent api-key store --resource lkc-xxxxx
Key: <API_KEY>
Secret: <API_SECRET>

API キーとシークレットがない場合は、以下で作成可能です：

confluent api-key create --resource lkc-xxxxx

追加後、以下で使用できます：

confluent api-key use "API_Key" --resource lkc-xxxxx

トピックの作成

confluent kafka topic create <topic-name>

トピック一覧の確認：

confluent kafka topic list

トピックへのメッセージ送信

confluent kafka topic produce <topic-name>

トピックからのメッセージ受信

confluent kafka topic consume -b <topic-name>

EMQX Cloud デプロイとの VPC ピアリング接続の確立

クラスター作成後、Confluent Cloud コンソールでピアリングを追加する必要があります。

1. Confluent Cloud 環境の Network Management タブに移動し、For dedicated cluster タブを選択します。接続したい特定のネットワークを選び、Ingress connections タブの下のボタンをクリックしてピアリングを追加します。

2. EMQX Cloud からの VPC 情報を入力します。（デプロイコンソールの VPC Peering セクションで取得可能です）

3. Confluent でピアリングが追加されたら、Network Overview ページに移動し、Confluent Cloud VPC ID をコピーします。

4. EMQX Cloud コンソールに戻り、コピーした Confluent Cloud VPC ID を入力して Confirm をクリックします。ステータスが Running に変われば、VPC ピアリング接続は正常に確立されています。

上記の手順で、プライベートネットワークの前提設定が完了します。

コネクターの作成

データ統合ルールを作成する前に、Kafka サーバーにアクセスするための Confluent コネクターを作成する必要があります。

1. デプロイメントに移動し、左ナビゲーションメニューから Data Integration をクリックします。
2. 初めてコネクターを作成する場合は、Data Forward カテゴリの下にある Confluent を選択します。すでにコネクターを作成済みの場合は、New Connector を選択し、続いて Data Forward カテゴリの下の Confluent を選択します。
3. New Connector ページで以下のオプションを設定します：
   • Connector Name：システムが自動的にコネクター名を生成します。
   • Bootstrap Hosts：Confluent クラスター設定ページのエンドポイントを入力します。EMQX 6.1.5 以降のデプロイメントでは、IPv6 アドレスは角括弧で囲みます（例：[2001:db8::1]:9092）。
   • IP Family：EMQX 6.1.5 以降のデプロイメントで、Confluent ブローカーに接続するための IP アドレスファミリーを選択します：
     • Auto (auto)：デフォルト。IP アドレスの場合はそのアドレスファミリーを使用し、ホスト名の場合は IPv4 を優先し、失敗時に IPv6 を試行します。
     • IPv4 (ipv4)：IPv4 のみで接続します。
     • IPv6 (ipv6)：IPv6 のみで接続します。
   • Username and Password：前述の Confluent Cloud CLI で作成した API キーとシークレットを入力します。
   • TLS Verify：EMQX 6.1.5 以降のデプロイメントで有効にすると、Confluent サーバー証明書の検証を行います。CA Cert に信頼された CA 証明書を設定してください。
   • その他のオプションはデフォルトのままか、ビジネスニーズに応じて設定してください。
4. Test ボタンをクリックします。Confluent サービスにアクセス可能であれば、成功メッセージが表示されます。
5. New ボタンをクリックして作成を完了します。

ルールの作成

次に、書き込むデータを指定し、処理済みデータを Confluent に転送するアクションをルールに追加するためのルールを作成します。

1. 新しいルールを作成し、SQL 入力欄に以下の SQL 文を入力します。本デモで使用するルールは、temp_hum/emqx トピックからメッセージを読み取り、JSON オブジェクトに client_id、topic、timestamp 情報を追加して拡充します。

   • up_timestamp：メッセージが報告された時間
   • client_id：メッセージをパブリッシュしたクライアントの ID
   • temp：メッセージペイロード内の温度データ
   • hum：メッセージペイロード内の湿度データ

   SELECT 
      timestamp as up_timestamp, 
      clientid as client_id, 
      payload.temp as temp,
      payload.hum as hum
      FROM
      "temp_hum/emqx"

2. テストペイロード、トピック、クライアント情報を入力して SQL ルールをテストし、SQL Test をクリックします。以下のような結果が表示されれば、SQL テストは成功です。

3. Next をクリックしてルールにアクションを追加します。ビジネスニーズに従って以下の設定項目を入力します：

   • Kafka Topic：testtopic-in と入力します。注意：ここでは変数はサポートされません。
   • Kafka Headers：Kafka メッセージに関連するメタデータやコンテキスト情報を入力します（任意）。プレースホルダーの値はオブジェクトである必要があります。Kafka Header Value Encoding Type ドロップダウンリストからヘッダー値のエンコードタイプを選択できます。Add をクリックしてさらにキー・バリューを追加可能です。
   • Message Key：Kafka メッセージのキーです。ここに文字列を入力できます。純粋な文字列またはプレースホルダー（${var}）を含む文字列が可能です。
   • Message Value：Kafka メッセージの値です。ここに文字列を入力できます。純粋な文字列またはプレースホルダー（${var}）を含む文字列が可能です。
   • Partition Strategy：プロデューサーがメッセージを Kafka のパーティションに分配する方法を選択します。
   • Compression：Kafka メッセージ内のレコードを圧縮／解凍するために圧縮アルゴリズムを使用するかどうかを指定します。

4. 詳細設定（任意）：高度な設定を参照してください。

5. Confirm ボタンをクリックして作成を完了します。作成後、ページはデータ統合の初期ページに戻ります。

これでルールの作成が完了し、Integration -> Rules ページに新規作成したルールが表示され、Actions タブには新規作成された action-xxx が確認できます。

ルールのテスト

Confluent プロデューサールールが期待通りに動作するかテストするために、MQTTX を使用してクライアントが EMQX に MQTT メッセージをパブリッシュする動作をシミュレートできます。

1. MQTTX を使って temp_hum/emqx トピックにメッセージを送信します。

   mqttx pub -i emqx_c -t t/1 -m '{"temp":"23.5","hum":"32.6"}'

2. データ統合ページで rule_id をクリックし、ルールおよびアクションの統計情報を確認します。マッチしたメッセージが1件、新たに通過したメッセージが1件あるはずです。

3. Confluent コンソールでデータを確認するか、以下の Confluent コマンドを使って emqx トピックにメッセージが書き込まれているか確認してください。

   confluent kafka topic consume -b emqx