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. 以下の情報を控えてください：

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

    event_hubs_dns_config

  6. これらの FQDN と IP アドレスを EMQX Cloud のサポートチケット経由で提出してください。EMQX Cloud の SRE チームがデプロイメントの DNS 解決設定を完了します。

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

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

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

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

  3. 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. テスト をクリックし接続確認。成功メッセージが表示されれば接続可能です。

  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 をクリックし、メッセージスループットやアクション実行結果などのランタイム統計を表示。

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