Skip to content

MS_AzureEventHubsTutorial

Jump to bottom
nishi_74322014 edited this page Sep 1, 2026 · 1 revision

Azure Event Hubsチュートリアル

概要

ここでは、C# で、IoT デバイス → Azure Event Hubs → Azure Databricksと
繋いでみる。

詳細

契約

Azure の契約

Event Hubs の契約

Basic レベル、Standard レベルを選択できるらしい。

補足（現在の SKU）: 現在は Basic / Standard / Premium / Dedicated の
4 段階である。後述のとおり Kafka エンドポイントは Basic では使えないため、
本チュートリアルの構成では Standard 以上が必要になる。
なお Basic は 2025 年に新規作成が停止されており、
新規構築では Standard が実質的な最小構成となる。

構築

取り敢えず、Kafkaエンドポイントを持つ Event Hubs を構築してみる。

Kafka エンドポイント

  • SKU

    • Basic レベルではサポートされていない。
    • Standard レベルを選択すれば利用可能である模様。

  • マッピング

Event Hubs Kafka
名前空間 クラスター
イベントハブ トピック
Partition Partition
コンシューマーグループ コンシューマーグループ
Offset Offset

  • 圧縮 / メッセージフォーマットは現在サポートしていない。

  • 認証機構として SASL 認証をサポートしている。
    が、その場合（≒ IoT デバイスから接続する場合）、
    推奨は、標準の接続文字列（SAS トークン）の利用らしい。

  • コンシューマ・グループは別モノ（完全に区別される）。

  • 追加の制限事項

    • group.id プロパティの最大長は 256 文字
    • offset.metadata.max.bytes の最大サイズは 1024 バイト

補足（Kafka API の互換範囲）: Event Hubs の Kafka エンドポイントは
Kafka のプロトコルを喋る Event Hubsであって、Kafka そのものではない。
このため、

  • ログ圧縮（log compaction）、トランザクション、
    AdminClient によるトピック作成といった一部の API は使えない。
  • パーティション数は作成後に増やせるが減らせない、
    保持期間は SKU の上限に従う、といった Event Hubs 側の制約が効く。

「Kafka のクライアント ライブラリをそのまま使える」ことが主眼で、
Kafka クラスタの完全な代替ではない点に注意する。

Event Hubs の作成

  • ポータルから

    • リソース グループを作成
      データ・パイプライン系は「West US 2」辺りが良いかも。

    • Event Hubs 名前空間を作成

    • Event Hubs を作成

移行メモ（誤字）: 元ページの「データ・パープライン系」は
「データ・パイプライン系」の誤りと解して修正した（2 箇所）。

  • IaC 化

    • Azure CLIで

      • リソース・グループがない場合、作成

        az group create --name [ResourceGroupName] --location [Location]
例：az group create --name EventHubsRG --location "Japan East"

      • Event Hubs 名前空間を作成

        az eventhubs namespace create --name [EventHubsNamespace] --resource-group [既存のRG名] -l [Location]
例：az eventhubs namespace create --name osscjpdevinfra --resource-group EventHubsRG -l "Japan East"

        ※ 省略してある --sku の既定値は Standard になっている。

      • イベント ハブの作成

        az eventhubs eventhub create --name [EventHubsName] --resource-group [既存のRG名] --namespace-name [EventHubs名前空間]
例：az eventhubs eventhub create --name OsscJpDevInfra --resource-group EventHubsRG --namespace-name osscjpdevinfra

      • IP アドレス制限

        az eventhubs namespace network-rule add --resource-group [ResourceGroupName] --namespace-name [EventHubs名前空間] --ip-address xxx.xxx.xxx.xxx/24 --action Allow
例：az eventhubs namespace network-rule add --resource-group EventHubsRG --namespace-name osscjpdevinfra --ip-address xxx.xxx.xxx.xxx/24 --action Allow

      • 参考
        ・https://opentouryo.osscons.jp/index.php?%E9%83%A8%E4%BC%9A%E3%83%A1%E3%83%A2%EF%BC%882021%E5%B9%B4%EF%BC%89
        ・https://github.com/OpenTouryoProject/DataPipeline/tree/develop/Gateway/EventHubs/IaC

移行メモ（脱字）: IP アドレス制限のコマンド例が
「--action」で終わっており値が欠けていたため、
直前の書式例に合わせて Allow を補った。

送受信

純正 Kafka クライアントで送受信

（純正 Kafka クライアントの導入は
https://dotnetdevelopmentinfrastructure.osscons.jp/index.php?Kafka%E3%83%81%E3%83%A5%E3%83%BC%E3%83%88%E3%83%AA%E3%82%A2%E3%83%AB を参照）

環境構築

  • 純正 Kafka クライアント
  • 上記「Event Hubs の作成」

接続文字列の取得

コマンドで送受信

WSL2で実行する。

  • 準備

    • 取得

      git clone https://github.com/Azure/azure-event-hubs-for-kafka.git
cd azure-event-hubs-for-kafka/quickstart/kafka-cli

    • 編集
      password 部分に primaryConnectionString 値を埋める。

      $ dir
README.md  client_common.properties  jaas.conf
$ vi jaas.conf

  • 送受信

    kafkaInstallHome=/usr/local/kafka
export KAFKA_OPTS="-Djava.security.auth.login.config=jaas.conf"
echo $kafkaInstallHome
echo $KAFKA_OPTS

    • 受信

      $kafkaInstallHome/bin/kafka-console-consumer.sh --topic test_topic --bootstrap-server $eventhubsNameSpace.servicebus.windows.net:9093 --consumer.config client_common.properties

    • 送信

      $kafkaInstallHome/bin/kafka-console-producer.sh --topic test_topic --broker-list $eventhubsNameSpace.servicebus.windows.net:9093 --producer.config client_common.properties

    ※ $eventhubsNameSpace は、上記「環境構築」で設定した値。

Kafka .NET クライアントで送受信

接続文字列の取得

ポータルの「共有アクセス ポリシー」から取得する。

Endpoint=sb://[EventHubsNamespace].servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

※ 内部的には、SAS トークンを使用する。

補足（RootManageSharedAccessKey は使わない）: 上記の例で使われている
RootManageSharedAccessKey は名前空間全体に対する管理権限を持つため、
デバイスやアプリに配るべきではない。
実運用では、

  • イベント ハブ単位で Send / Listen だけを持つポリシーを作る、
  • あるいは接続文字列を配らず Microsoft Entra ID のマネージド IDで
    接続する（Azure Event Hubs Data Sender などのロールを割り当てる）

のが現在の推奨である。

サンプルをクローンして実行

  • 設定値

    • 環境に合わせて設定する。

      • EH_FQDN
      • EH_CONNECTION_STRING

    • 必要に応じて設定する。

    • CA_CERT_LOCATION

      • そのままでも動作する（また、設定しなくても動作する）
      • パラメタとしては、Confluent.Kafka の SslCaLocation。
      • Azure Event Hubs の Kafka エンドポイントは既定で SSL らしい。
      • 必要であれば、このパラメタで、認証局の証明書をポイントする。

  • 実行結果の例

    Initializing Producer
Sending 10 messages to topic: test, broker(s): osscjpdevinfra.servicebus.windows.net:9093
Message 0 sent (value: 'Sample message #0 sent at 2021-07-23_18:19:22.9101')
Message 1 sent (value: 'Sample message #1 sent at 2021-07-23_18:19:23.3055')
Message 2 sent (value: 'Sample message #2 sent at 2021-07-23_18:19:23.3599')
Message 3 sent (value: 'Sample message #3 sent at 2021-07-23_18:19:23.4080')
Message 4 sent (value: 'Sample message #4 sent at 2021-07-23_18:19:23.4652')
Message 5 sent (value: 'Sample message #5 sent at 2021-07-23_18:19:23.5150')
Message 6 sent (value: 'Sample message #6 sent at 2021-07-23_18:19:23.5784')
Message 7 sent (value: 'Sample message #7 sent at 2021-07-23_18:19:23.6300')
Message 8 sent (value: 'Sample message #8 sent at 2021-07-23_18:19:23.6865')
Message 9 sent (value: 'Sample message #9 sent at 2021-07-23_18:19:23.7695')

Initializing Consumer
Consuming messages from topic: test, broker(s): osscjpdevinfra.servicebus.windows.net:9093
Received: 'Sample message #0 sent at 2021-07-23_18:19:22.9101'
Received: 'Sample message #1 sent at 2021-07-23_18:19:23.3055'
Received: 'Sample message #2 sent at 2021-07-23_18:19:23.3599'
Received: 'Sample message #3 sent at 2021-07-23_18:19:23.4080'
Received: 'Sample message #4 sent at 2021-07-23_18:19:23.4652'
Received: 'Sample message #5 sent at 2021-07-23_18:19:23.5150'
Received: 'Sample message #6 sent at 2021-07-23_18:19:23.5784'
Received: 'Sample message #7 sent at 2021-07-23_18:19:23.6300'
Received: 'Sample message #8 sent at 2021-07-23_18:19:23.6865'
Received: 'Sample message #9 sent at 2021-07-23_18:19:23.7695'

Apache Spark 系

.NET for Apache Spark で送受信

Scala で送受信

PySpark で送受信

Azure Databricksから接続

上記の受信処理を
Azure Databricks上で実行する。

参考

Qiita

https://qiita.com/tags/eventhubs

GitHub

confluent-kafka-dotnet

Azure Event Hubs for Apache Kafka Ecosystems

https://github.com/Azure/azure-event-hubs-for-kafka

Microsoft Docs

クイック スタート

チュートリアル

操作方法ガイド

ナレコム Azure レシピ

https://azure-recipe.kc-cloud.jp/category/azure-event-hubs/

概要とパフォーマンス

触ってみた

移行メモ（重複）: 「③ メッセージ送受信編」が元ページで
2 回書かれていたため、1 つにまとめた。

本 Wiki 内

Tags: クラウド, ビッグデータ, Azure

このWikiは「Open棟梁Project」,「OSSコンソーシアム 開発基盤部会」によって運営されています。

NetDevInfraWiki

マイクロソフト系技術情報 Wiki
Open 棟梁 Wiki

（未着手）

開発基盤部会 Wiki

移行管理: DONE ／ TODO

Clone this wiki locally