Kafka から Azure Stream Analytics へのデータのストリーム

Apache Kafkaは、アプリが記録のストリームを公開・購読するために利用するオープンソースの分散型ストリーミングプラットフォームです。 メッセージング、ウェブサイトの活動追跡、指標、ログ集約、ストリーム処理などに一般的に使われています。

もしデータがKafkaクラスターを経由しているなら、Azure Stream Analyticsを直接そのクラスターに接続し、カスタムコンシューマーを作らずにデータを取り込み処理できます。 Kafkaの入力はローコードで、ビジネスコンプライアンス基準を満たすためにMicrosoftによって完全に管理され、0.10以降のクライアントバージョンとも後方互換性があります。 既存のKafka設定規則を用いて、仮想ネットワーク内またはパブリックエンドポイントでKafkaクラスタに接続できます。 サポートされている圧縮の種類は、None、Gzip、Snappy、LZ4、Zstd です。

この記事では、Azure Stream Analyticsジョブの入力ソースとしてKafkaクラスタを設定する方法を紹介します。

カフカを入力ソースに設定する

この記事では、Azure Stream Analytics の入力ソースとして Kafka を設定する方法について説明します。 次の手順について説明します。

1. Azure Stream Analytics ジョブを作成します。
2. mTLS または SASL_SSL セキュリティ プロトコルを使用している場合は、マネージド ID を使用するように Azure Stream Analytics ジョブを構成します。
3. mTLSやSASL_SSLセキュリティプロトコルを使っている場合はAzure Key Vault設定してください。
4. 証明書を秘密としてAzure Key Vaultにアップロードします。
5. アップロードされた証明書にアクセスするためのアクセス許可を Azure Stream Analytics に付与します。
6. Azure Stream Analytics ジョブで Kafka 入力を構成します。

Kafka入力の設定

Kafka入力を設定して、Azure Stream AnalyticsジョブがKafkaクラスタに接続してトピックから読み取れるようにしてください。

1. Azure Stream Analyticsのジョブで、Kafka入力を追加してください。

2. 各入力プロパティは、以下の表に示された値を用いて設定します。

   プロパティ名	説明
   入出力エイリアス	入力や出力を指すためにクエリで使われる親しみやすい名前です。
   ブートストラップ サーバー アドレス	Kafkaクラスタに接続するためのホスト/ポートペアのリストです。
   Kafka トピック	メッセージの公開・購読およびイベント駆動型処理をサポートする、名前付き、順序付けられ、パーティション化されたデータストリームです。
   セキュリティ プロトコル	Kafka クラスターへの接続方法。 Azure Stream Analytics は、mTLS、SASL_SSL、SASL_PLAINTEXT、または None をサポートしています。
   コンシューマー グループ ID	入力が属する必要がある Kafka コンシューマー グループの名前。 Azure Stream Analyticsは、提供しなければ自動的に割り当ててくれます。
   イベント シリアル化形式	受信データストリームのシリアライゼーション形式(JSON、CSV、Avro、Parquet、Protobuf)。

3. 入力設定を保存してください。

認証と暗号化の設定

認証と暗号化を設定し、Azure Stream AnalyticsジョブとKafkaクラスタ間の接続を安全に保ちましょう。 Confluent Cloud Kafkaへの接続に関するステップバイステップのチュートリアルについては、「Stream data from Confluent Cloud Kafka with Azure Stream Analytics」の入力、出力の「Stream data from Azure Stream Analytics into Confluent Cloud」をご覧ください。

1. 以下の表に説明されているように、あなたのKafkaクラスターに合った4つのサポートされたセキュリティプロトコルの中から1つを選択します。

   プロパティ名	説明
   mTLS	暗号化および認証。 PLAIN、SCRAM-SHA-256、および SCRAM-SHA-512 のセキュリティ メカニズムをサポートします。
   SASL_SSL	SASL(シンプル認証およびセキュリティ層)とセキュアソケット層(SSL)という2つの異なるセキュリティ機構を組み合わせ、データ伝送のために認証と暗号化の両方が実施されていることを保証します。 SASL_SSL プロトコルは、PLAIN、SCRAM-SHA-256、および SCRAM-SHA-512 セキュリティ メカニズムをサポートします。
   SASL_PLAINTEXT	ユーザー名は標準認証ですが暗号化はなしです。
   なし	認証と暗号化はありません。

2. Azure Stream Analyticsジョブで選択したプロトコルの認証情報を設定してください。

Azure Key Vault との統合

Azure Stream AnalyticsはAzure Key Vaultと連携し、mTLSやSASL_SSLセキュリティプロトコルを使う際に認証や暗号化に必要な保存された秘密にアクセスします。 あなたのAzure Stream Analyticsジョブは、管理IDを使ってAzure Key Vaultに接続し、安全な接続を確保し、秘密の流出を防ぎます。 証明書はシークレットとしてキー コンテナーに格納され、PEM 形式である必要があります。 Azure Stream AnalyticsのジョブをAzure Key Vaultと統合するには、以下のタスクを行ってください。以下のセクションで詳細に説明されています。

1. キーボールトに必要な権限を設定してください。
2. 証明書をキーボールトにシークレットとしてアップロードしてください。
3. Azure Stream AnalyticsのジョブでマネージドIDを設定してください。
4. Stream Analyticsのジョブにキーボールト内の証明書へのアクセス権限を与えてください。

権限を設定してキー ボールトを構成する

キーボールトリソースを作成するには、「クイックスタート:Azureポータルを使ってキーボールトを作成する」をご覧ください。 証明書をアップロードするにはkey vault Key Vault管理者のアクセスが必要です。 管理者権限を付与するには、以下の手順に従ってください:

1. [アクセス制御 (IAM)] を選択します。

2. [追加]>[ロールの割り当ての追加] を選択して、[ロールの割り当ての追加] ページを開きます。

3. 以下の構成を使って役割を割り当てます:

   設定	値
   役割	キーボールト管理者
   アクセスを割り当てる	ユーザー、グループ、またはサービス プリンシパル
   メンバー	<アカウント情報またはメール アドレス>

証明書をキーボールトにアップロードしてください

Azure CLIはPowerShellでローカルで設定されていることを確認してください。 Azure CLIの設定に関するガイダンスについては、「Azure CLIで始めて」をご覧ください。

以下の手順で証明書を秘密としてキーボールトにアップロードします。

1. Azure CLIにサインインします。

   az login
   

2. 鍵の保管庫を含むサブスクリプションに接続してください。

   az account set --subscription <subscription name>
   

3. 証明書をシークレットとしてアップロードしてください。 <your key vault>をキーボールトの名前に置き換え、秘密を付けたい名前に<name of the secret>、証明書ファイルへのパスに<file path to certificate>してください。 右クリックして、証明書へのパスをコピーできます。

   az keyvault secret set --vault-name <your key vault> --name <name of the secret> --file <file path to certificate>
   

   次に例を示します。

   az keyvault secret set --vault-name mykeyvault --name kafkasecret --file C:\Users\Downloads\certificatefile.pem
   

マネージド ID を構成する

Azure Stream AnalyticsはキーボールトにアクセスするためにマネージドIDが必要です。 Azure Stream Analyticsジョブを管理IDを使用するように設定するには、「Configure」の管理付きIDタブに移動します。

マネージドIDを設定するには、以下の手順に従ってください:

1. 設定の下にある管理されたアイデンティティタブを選択します。

   

2. スイッチ 識別子を選択し、ジョブで使用する識別子を選択します:システム割り当て識別子またはユーザー割り当て識別子。

3. ユーザー割り当てIDについては、ユーザー割り当てIDがあるサブスクリプションを選択し、そのID名を選択してください。

4. レビューして「 保存」を選択してください。

Stream Analyticsのジョブ権限を付与して証明書にアクセスしてください

Azure Stream Analytics ジョブでキー コンテナー内のシークレットを読み取る場合、ジョブにはキー コンテナーにアクセスするためのアクセス許可が必要です。

Stream Analyticsの職に権限を与えるには、以下の手順に従ってください:

1. [アクセス制御 (IAM)] を選択します。

2. [追加]>[ロールの割り当ての追加] を選択して、[ロールの割り当ての追加] ページを開きます。

3. 以下の構成を使って役割を割り当てます:

   設定	値
   役割	Key Vault 秘密ユーザー
   マネージド ID	システム割り当てマネージド ID またはユーザー割り当てマネージド ID の Stream Analytics ジョブ
   メンバー	<Stream Analytics ジョブの名前> または <ユーザー割り当て ID の名前>

カフカ入力の制限と考慮事項

Kafkaクラスターが仮想ネットワーク内またはファイアウォールの背後にある場合は、プライベートリンクや専用ネットワーク設定を使ってAzure Stream AnalyticsジョブにKafkaトピックにアクセスするように設定してください。 詳細については、「Run your Azure Stream Analytics job in an Azure Virtual Network」をご覧ください。

カフカを入力ソースとして使用する際の以下の制限を考慮してください。

1. Azure Stream AnalyticsジョブをVirtual Network/SWIFT使用に設定する際は、少なくとも6台のストリーミングユニットか1台のV2ストリーミングユニットで設定してください。
2. mTLSやSASL_SSLをAzure Key Vaultで使用する場合、JavaキーストアをPEM形式に変換する必要があります。
3. Azure Stream Analyticsが接続できるKafkaの最低バージョンはバージョン0.10です。
4. Azure Stream Analyticsは、OAuthやSAMLシングルサインオン(SSO)によるConfluent Cloudへの認証をサポートしていません。 SASL_SSLプロトコルを通じてAPIキーを使用する必要があります。

Azure Stream Analytics の Kafka 入力について直接サポートが必要な場合は、askasa@microsoft.com にお問い合わせください。

関連するコンテンツ

• クイック スタート: Azure Portal を使用して Stream Analytics ジョブを作成する
• Azure Stream Analyticsを使用して Confluent Cloud Kafka からデータをストリーミングする
• Azure Stream AnalyticsからConfluent CloudへデータをStream