EMQX Cloud

日本語

简体中文
English

日本語

简体中文
English

Appearance

お問い合わせ
お問い合わせ

Azure Event Hubs に MQTT データをストリームする ​

Azure Event Hubs は、マイクロソフトが提供するフルマネージドのイベントストリーミングサービスで、大規模なリアルタイムデータ取り込みに対応しています。EMQX Cloud は Azure Event Hubs と連携し、MQTT データを Azure エコシステムへ信頼性高く高スループットでストリーム配信します。

本ドキュメントでは、EMQX Cloud と Azure Event Hubs の連携方法を紹介します。コネクターの作成、ルールの定義、連携のテストを、MQTT クライアントからシミュレートした温度・湿度データを Azure Event Hubs にストリーム配信する簡単な例を用いて説明します。

動作の仕組み ​

EMQX Cloud は Kafka 互換のエンドポイントを介して Azure Event Hubs と連携し、TLS 上の SASL/PLAIN 認証を利用します。この連携は EMQX のルールエンジンとシンク機構を活用し、MQTT データを安全かつ効率的に Azure エコシステムへストリーム配信します。

エンドツーエンドのワークフローは以下の通りです：

  1. デバイスが MQTT メッセージをパブリッシュ：IoT デバイスが EMQX Cloud 上の特定の MQTT トピックにテレメトリデータをパブリッシュします。
  2. ルールエンジンがデータを処理：ルールエンジンはトピックマッチングや SQL ルールに基づき、受信した MQTT メッセージをフィルタリングおよび変換します。データの抽出、再フォーマット、追加コンテキストの付与が可能です。
  3. データを Azure Event Hubs に転送：マッチしたルールはアクションをトリガーし、処理済みメッセージを Kafka プロトコル経由で Azure Event Hubs に送信します。MQTT トピックやユーザープロパティは Event Hubs メッセージヘッダーにマッピングでき、必要に応じてメッセージキーを設定し順序制御が可能です。

Azure Event Hubs にデータが取り込まれると、Stream Analytics、Blob Storage、Azure 仮想マシン上のカスタムアプリケーションなどの下流 Azure サービスでシームレスに消費でき、リアルタイム処理、ストレージ、イベント駆動型アプリケーション開発が可能になります。

特長と利点 ​

  • 高性能なメッセージスループット

    EMQX Cloud は低レイテンシで毎秒数百万件の MQTT メッセージを Azure Event Hubs に取り込み、IoT の高スループットワークロードをサポートします。

  • 柔軟なデータマッピング

    MQTT トピックやユーザープロパティを Azure Event Hubs のイベントやヘッダーに柔軟にマッピングでき、豊富なコンテキスト付与やメッセージ順序の制御が可能です。

  • 弾力的なスケーラビリティ

    EMQX Cloud と Azure Event Hubs は共にメガバイトからテラバイト規模のデータ量に対応可能で、アーキテクチャの変更なしにスケールアウトできます。

  • 豊富な Azure と Kafka エコシステム連携

    MQTT デバイス接続と Azure Event Hubs の Azure サービス、SDK、Kafka エコシステムのサポートを組み合わせることで、シームレスなエンドツーエンドの IoT データ取り込みと処理を実現します。

はじめる前に ​

このセクションでは、データ統合の設定前に必要な前提条件と準備について説明します。EMQX Cloud でコネクターやルールを作成する前に、ネットワーク設定、Azure リソース、アクセス資格情報が適切に準備されていることを確認してください。

前提条件 ​

ネットワークの設定 ​

データ統合を構成する前に、EMQX Cloudのデプロイメントを作成し、EMQX Cloudと対象サービス間のネットワーク接続を確立していることを確認してください。

  • Dedicated Flexデプロイメントの場合：

    EMQX CloudのVPCと対象サービスのVPC間でVPCピアリング接続を作成します。ピアリング接続が確立されると、EMQX Cloudは対象サービスのプライベートIPアドレスを介してアクセス可能になります。

    パブリックIP経由でのアクセスが必要な場合は、NATゲートウェイを構成してアウトバウンド接続を有効にしてください。

  • BYOC（Bring Your Own Cloud）デプロイメントの場合：

    BYOCデプロイメントが稼働しているVPCと対象サービスをホストするVPC間でVPCピアリング接続を作成します。ピアリングが確立されると、対象サービスのプライベートIPアドレスを介してアクセス可能になります。

    対象サービスにパブリックIP経由でアクセスする必要がある場合は、クラウドプロバイダーのコンソールを使用してBYOC VPCにNATゲートウェイを構成してください。

Azure Event Hubs ネームスペースの設定 ​

EMQX Cloud からデータを受け取るために Azure Event Hubs ネームスペースを作成する必要があります。既にネームスペースをお持ちの場合は、このステップをスキップしてプライベートエンドポイントの作成に進んでください。

  1. Azure ポータルで Event Hubs に移動し、作成 を選択します。

  2. 基本 タブで以下を設定します：

    • サブスクリプション：Azure サブスクリプションを選択します。

    • リソースグループ：既存のリソースグループを選択するか、新規作成します。

    • ネームスペース名：ネームスペースの一意な名前を入力します。

    • リージョン：EMQX Cloud のデプロイと同じリージョンを選択します。

    • 価格レベル：Standard, Premium, または Dedicated を選択します。

      Basic レベルは Apache Kafka をサポートしておらず、使用できません。

    event_hubs_namespace

  3. 次へ: 詳細設定 をクリックし、デフォルト設定のまま進みます。

  4. 次へ: ネットワーク をクリックして続行します。

プライベートエンドポイントの作成 ​

EMQX Cloud が Azure Event Hubs に安全にアクセスできるよう、ネームスペースにプライベートエンドポイントを設定する必要があります。

  • 新規ネームスペース作成時は、ネットワーク タブで接続方法として プライベートアクセス を選択し、+ プライベートエンドポイント をクリックします。
  • 既存ネームスペースの場合は、概要 -> ネットワーク -> プライベートアクセス -> + プライベートエンドポイント に進みます。

  1. プライベートエンドポイントの作成 パネルで以下を設定します：

    • サブスクリプション と リソースグループ を選択します。
    • ターゲットサブリソース を namespace に設定します。
    • プライベート接続に使用する 仮想ネットワーク と サブネット を選択します。

  2. プライベート DNS 統合 では、プライベート DNS ゾーンと統合する を はい のままにし、デフォルトのプライベート DNS ゾーンを使用します。

  3. OK をクリックしてプライベートエンドポイントを作成します。

    event_hubs_private_endpoint

  4. プライベートエンドポイントが作成され、ステータスが Succeeded になったら、プライベートエンドポイント -> DNS 構成 に移動します。

  5. 以下の情報を記録してください：

    • プライベート DNS FQDN（例：xxx.privatelink.servicebus.windows.net）
    • プライベート IP アドレス

    event_hubs_dns_config

  6. privatelink.servicebus.windows.net で終わるプライベート FQDN と対応するプライベート IP アドレスのみを記載したサポートチケットを提出してください。EMQX Cloud の SRE チームが、FQDN とプライベート IP アドレスをマッピングする DNS レコードをデプロイメント向けに作成します。

Azure Event Hubs とアクセスポリシーの設定 ​

ネームスペースが準備できたら、Event Hub を作成し、EMQX Cloud で使用する接続文字列を取得します。

  1. Event Hubs ネームスペース ページで + Event Hub をクリックします。

  2. Event Hub 名を入力し、その他の設定はデフォルトのままにして 確認および作成 をクリックします。

  3. Event Hub 作成後、対象の Event Hub を開き、設定 -> 共有アクセス ポリシー に移動します。

  4. + 追加 をクリックし、ポリシー名を入力し、必要な権限（例：送信 または 管理）を割り当てます。

  5. ポリシーを保存し、プライマリ接続文字列 をコピーします。

    event_hunbs_sas_policies

この接続文字列は後で EMQX Cloud で Azure Event Hubs コネクターを作成する際に使用します。

コネクターの作成 ​

Azure Event Hubs は Kafka 互換のエンドポイントを提供しており、EMQX Cloud は Apache Kafka プロトコルを使ってデータをパブリッシュできます。MQTT データを Azure Event Hubs に転送するには、まず EMQX Cloud で Azure Event Hubs コネクターを作成する必要があります。

事前に Event Hubs ネームスペースの共有アクセスポリシーから プライマリ接続文字列 を取得しておいてください。

  1. デプロイメントに移動し、左側メニューの データ統合 をクリックします。

  2. コネクターを作成します：

    • コネクターが存在しない場合は、データ転送 カテゴリの中から Azure Event Hubs を選択します。
    • 既にコネクターがある場合は、新しいコネクター をクリックし、データ転送 の中から Azure Event Hubs を選択します。

  3. 新しいコネクター ページで以下を設定します：

    • コネクター名：自動生成されます。必要に応じて変更してください。
    • Bootstrap Hosts：ネームスペースのホスト名を入力します（例：xxx.servicebus.windows.net:9093）。EMQX 6.1.5 以降のデプロイメントでは、IPv6 アドレスの場合は角括弧で囲みます（例：[2001:db8::1]:9093）。
    • IP ファミリー：EMQX 6.1.5 以降のデプロイメントで Azure Event Hubs への接続に使用する IP アドレスファミリーを選択します。
      • Auto (auto)：デフォルト。IP アドレスの場合はそのファミリーを使用し、ホスト名の場合は IPv4 を優先し、失敗時に IPv6 を試みます。
      • IPv4 (ipv4)：IPv4 のみで接続します。
      • IPv6 (ipv6)：IPv6 のみで接続します。
    • 接続文字列：ネームスペースの共有アクセスポリシーからコピーした プライマリ接続文字列 を貼り付けます。
    • TLS を有効にする：デフォルトで有効。Azure Event Hubs への接続に必須です。

  4. テスト をクリックして接続を検証します。成功メッセージが表示されれば、EMQX Cloud から Azure Event Hubs へのアクセスが可能です。

  5. 新規作成 をクリックしてコネクターを作成します。

作成したコネクターは複数のルールで再利用でき、MQTT データを Azure Event Hubs に転送できます。

ルールの作成 ​

Azure Event Hubs コネクターを作成したら、どの MQTT データを転送し、どのように Azure Event Hubs に書き込むかを定義するルールを作成します。

  1. 新しいルールを作成します：

    • データ統合 -> ルール で 新しいルール をクリックするか、
    • Azure Event Hubs コネクターの アクション 列にある 新しいルール アイコンをクリックします。

  2. SQL エディター に、受信 MQTT メッセージをマッチング・処理するルールを入力します。

    以下の例は、temp_hum/emqx トピックのメッセージからタイムスタンプ、クライアント ID、温度、湿度の値を抽出します：

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

    Try It Out を使ってデータ入力をシミュレートし、クエリ結果を検証できます。

  3. 次へ をクリックしてアクションを追加します。

  4. アクション 設定で以下を指定します：

    • コネクター：作成済みの Azure Event Hubs コネクターを選択します。

    • Event Hub 名：転送先の Event Hub 名を入力します（変数はサポートされません）。

    • Azure Event Hub ヘッダー：MQTT データを Event Hubs メッセージヘッダーにマッピングするプレースホルダーを指定します。

    • Azure Event Hub ヘッダー値のエンコードモード：ヘッダー値のエンコード方法を選択します（例：none または json）。

    • 追加の Azure Event Hub ヘッダー：（任意）追加のキー・バリュー形式ヘッダーを設定します。

    • メッセージキー：固定文字列またはプレースホルダー（例：${client_id}）を使ってメッセージキーを定義し、メッセージの順序制御を行います。

    • メッセージ値：メッセージペイロードを定義します。抽出したフィールドをプレースホルダーで挿入可能です：

      json
      {"temp": ${temp}, "hum": ${hum}}

    • メッセージタイムスタンプ：Event Hubs メッセージに使用するタイムスタンプの種類を選択します。

  5. （任意）詳細設定 を構成します。例：

    • 最大バッチサイズ（バイト）
    • 必要なアック数
    • パーティション戦略

  6. 確定 をクリックしてルールを作成します。

  7. 成功ダイアログで ルールに戻る をクリックして設定を完了します。

ルールが有効になると、マッチした MQTT メッセージが自動的に処理され、Azure Event Hubs に転送されます。

ルールのテスト ​

ルール作成後、MQTT メッセージをパブリッシュして Azure Event Hubs への配信を確認し、データ統合を検証できます。

テストデータのパブリッシュ ​

任意の MQTT クライアントを使ってテストメッセージをパブリッシュできます。以下は MQTTX を使った例です。

  1. MQTTX を EMQX Cloud デプロイメントに接続します。

  2. 以下のトピックにメッセージをパブリッシュします：

    • トピック：temp_hum/emqx

    • ペイロード：

      json
      {
  "temp": "27.5",
  "hum": "41.8"
}

Azure Event Hubs でのデータ確認 ​

  1. Kafka 互換のコンシューマーを使い、Event Hub からメッセージを読み取ります。

  2. Azure GitHub リポジトリにある Python コンシューマーなど、公式の Azure Event Hubs Kafka サンプルの使用を推奨します：

    bash
    python consumer.py <consumer-group> <topic.1> <topic.2> ... <topic.n>

    これにより、EMQX Cloud にパブリッシュされたメッセージが Kafka エンドポイント経由で正常に受信されていることを検証できます。

Kafka CLI を使ったメッセージ消費の詳細は、Azure ドキュメントを参照してください。

ルール実行のモニタリング ​

EMQX Cloud コンソールでルールの実行状況も確認できます：

  1. データ統合 -> ルール に移動します。

  2. ルール ID をクリックし、メッセージスループットやアクション実行結果などのランタイム統計を表示します。

メッセージの正常配信とルールメトリクスの増加が確認できれば、データ統合が期待通りに動作していることを示します。