クイック スタート: Azure Event Hubs と Apache Kafka を使用してデータをストリーミングする

このクイックスタートでは、Apache Kafka プロトコルを使用して Azure Event Hubs との間でデータをストリーミングする方法について説明します。 サンプルの Kafka プロデューサー アプリやコンシューマー アプリのコードは変更しません。 Event Hubs 名前空間を指すためにクライアントが使用する構成を更新するだけで、Kafka エンドポイントが公開されます。 また、独自に Kafka クラスターをビルドして使用することもありません。 代わりに、Kafka エンドポイントで Event Hubs 名前空間を使用します。

Note

このサンプルは GitHub で入手できます。

前提条件

このクイック スタートを完了するには、次の前提条件を満たしている必要があります。

• Apache Kafka 用の Event Hubs に関する記事を読む。
• Azure サブスクリプション。 お持ちでない場合は、開始する前に無料アカウントを作成してください。
• Windows 仮想マシンを作成し、次のコンポーネントをインストールします。
  • Java Development Kit (JDK) 1.7 以降
  • Maven バイナリ アーカイブのダウンロードとインストール
  • Git

Azure Event Hubs 名前空間を作成する

Event Hubs 名前空間を作成すると、名前空間の Kafka エンドポイントが自動的に有効になります。 Kafka プロトコルが使用されているアプリケーションからイベント ハブにイベントをストリーミングできます。 Event Hubs 名前空間を作成するには、Azure portal を使用したイベント ハブの作成に関するページの手順に従います。 専用クラスターを使用している場合は、専用クラスターでの名前空間とイベント ハブの作成に関する記事を参照してください。

Note

Kafka の Event Hubs は、Basic レベルではサポートされていません。

Event Hubs で Kafka を使用してメッセージを送受信する

パスワードレス (推奨)

1. 仮想マシンのシステム割り当てマネージド ID を有効にします。 VM でのマネージド ID の構成について、詳しくは「Azure portal を使用して Azure VM で Azure リソースのマネージド ID を構成する」を参照してください。 Azure リソースのマネージド ID は、Microsoft Entra ID で自動的に管理される ID を Azure サービスに提供します。 この ID を使用すると、コード内に資格情報を記述することなく、Microsoft Entra の認証をサポートする任意のサービスに対して認証を行うことができます。

   

2. 作成した Event Hubs 名前空間の [アクセス制御] ページを使用して、"Azure Event Hubs データ所有者" のロールを VM のマネージド ID に割り当てます。 Azure Event Hubs は、Microsoft Entra ID を使用して Event Hubs リソースへの要求を承認することをサポートしています。 Microsoft Entra ID では、Azure ロールベースのアクセス制御 (Azure RBAC) を使用して、セキュリティ プリンシパル (ユーザー、またはアプリケーションのサービス プリンシパルである可能性があります) にアクセス許可を付与できます。

   1. Azure portal で Event Hubs 名前空間に移動します。 左側のナビゲーションで [アクセス制御 (IAM)] に移動します。

   2. [+ 追加]、Add role assignment の順に選択します。

      

   3. [ロール] タブで、[Azure Event Hubs データ所有者] を選択して [次へ] ボタンを選択します。

      

   4. [メンバー] タブの [アクセス権の割り当て先] セクションで [マネージド ID] を選択します。

   5. [+ メンバーの選択] リンクを選択します。

   6. [マネージド ID の選択] ページで、次の手順に従います。

      1. VM がある Azure サブスクリプションを選択します。

      2. [マネージド ID] で、[仮想マシン] を選択します。

      3. 仮想マシンのマネージド ID を選択します。

      4. ページの下部にある [選択] を選びます。

         

   7. [レビューと割り当て] を選択します。

      

3. VM を再起動し、マネージド ID を構成した VM にサインインし直します。

4. Kafka 用 Azure Event Hubs リポジトリを複製します。

5. azure-event-hubs-for-kafka/tutorials/oauth/java/managedidentity/consumer に移動します。

6. src/main/resources/ フォルダーに切り替えて consumer.config を開きます。 namespacename は、実際の Event Hubs 名前空間の名前に置き換えてください。

   bootstrap.servers=NAMESPACENAME.servicebus.windows.net:9093
   security.protocol=SASL_SSL
   sasl.mechanism=OAUTHBEARER
   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
   sasl.login.callback.handler.class=CustomAuthenticateCallbackHandler;
   

   Note

   Kafka 用 Event Hubs の OAuth サンプルはすべてこちらにあります。

7. pom.xml ファイルがある Consumer フォルダーに戻り、Kafka クライアントを使用して、コンシューマー コードを実行してイベント ハブからのイベントを処理します。

   mvn clean package
   mvn exec:java -Dexec.mainClass="TestConsumer"                                    
   

8. 別のコマンド プロンプト ウィンドウを起動し、azure-event-hubs-for-kafka/tutorials/oauth/java/managedidentity/producer に移動します。

9. src/main/resources/ フォルダーに切り替えて producer.config を開きます。 mynamespace は、実際の Event Hubs 名前空間の名前に置き換えてください。

10. pom.xml ファイルがある Producer フォルダーに戻り、プロデューサー コードを実行して、イベントを Event Hubs にストリーミングします。

    mvn clean package
    mvn exec:java -Dexec.mainClass="TestProducer"                                    
    

    プロデューサー ウィンドウに送信されたイベントに関するメッセージが表示されます。 ここで、コンシューマー アプリ ウィンドウに切り替えて、イベント ハブから受信したメッセージを確認します。

    

接続文字列

1. Kafka 用 Azure Event Hubs リポジトリを複製します。

2. azure-event-hubs-for-kafka/quickstart/java/consumer に移動します。

3. src/main/resources/consumer.config にあるコンシューマーの構成の詳細を次のように更新します。

   bootstrap.servers=NAMESPACENAME.servicebus.windows.net:9093
   security.protocol=SASL_SSL
   sasl.mechanism=PLAIN
   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{YOUR.EVENTHUBS.CONNECTION.STRING}";
   

   重要

   {YOUR.EVENTHUBS.CONNECTION.STRING} を Event Hubs 名前空間への接続文字列に置き換えます。 接続文字列を取得する手順については、「Event Hubs の接続文字列の取得」を参照してください。 構成の例には、sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://mynamespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXX"; などがあります。

4. Kafka クライアントを使用してイベント ハブからコンシューマー コードを実行し、イベントを処理します。

   mvn clean package
   mvn exec:java -Dexec.mainClass="TestConsumer"
   

5. azure-event-hubs-for-kafka/quickstart/java/producer に移動します。

6. src/main/resources/producer.config にあるプロデューサーの構成の詳細を次のように更新します。

   bootstrap.servers=NAMESPACENAME.servicebus.windows.net:9093
   security.protocol=SASL_SSL
   sasl.mechanism=PLAIN
   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{YOUR.EVENTHUBS.CONNECTION.STRING}";
   

   重要

   {YOUR.EVENTHUBS.CONNECTION.STRING} を Event Hubs 名前空間への接続文字列に置き換えます。 接続文字列を取得する手順については、「Event Hubs の接続文字列の取得」を参照してください。 構成の例には、sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://mynamespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXX"; などがあります。

7. プロデューサー コードを実行し、Event Hubs にイベントをストリーム配信します。

   mvn clean package
   mvn exec:java -Dexec.mainClass="TestProducer"
   

Event Hubs Kafka クラスターにイベントがある場合は、コンシューマーからそれらの受信を開始します。

スキーマ レジストリを使用した Kafka のスキーマ検証

Event Hubs を使用して Kafka アプリケーションでデータをストリーミングする場合、Azure スキーマ レジストリを使用してスキーマを検証できます。 Event Hubs の Azure スキーマ レジストリにはスキーマ管理用に一元化されたリポジトリが用意されており、新規または既存の Kafka アプリケーションとスキーマ レジストリをシームレスに接続できます。

詳しくは、「Avro を使用して Apache Kafka アプリケーションのスキーマを検証する」をご覧ください。

次のステップ

この記事では、プロトコル クライアントを変更したり独自のクラスターを実行したりせずに Event Hubs にストリーム配信する方法を紹介しました。 詳細については、「Azure Event Hubs のための Apache Kafka 開発者ガイド」を参照してください。