apache Kafka ソースを Fabric イベントストリームに追加する

この記事では、Apache Kafka ソースを Fabric イベントストリームに追加する方法について説明します。

Apache Kafka は、スケーラブルでリアルタイムのデータ システムを構築するためのオープンソースの分散プラットフォームです。 Apache Kafka をイベントストリーム内のソースとして統合することで、Apache Kafka からリアルタイム イベントをシームレスに取り込み、それらを処理してから、Fabric内の複数の宛先にルーティングできます。

前提条件

• 共同作成者以上のアクセス許可を持つFabric ワークスペースへのアクセス。

• 稼働中の Apache Kafka クラスター。

• Apache Kafka にはパブリックにアクセスできる必要があり、ファイアウォールの内側にあったり、仮想ネットワークで保護されていたりしてはなりません。 保護されたネットワークに存在する場合は、 Eventstream コネクタの仮想ネットワークインジェクションを使用して接続します。

• TLS/mTLS 設定を使用する場合は、Azure Key Vault。

  • 必要な証明書を .pem 形式でAzure Key Vaultにインポートします。
  • ソース データとプレビュー データを構成するユーザーには、Key Vault内の証明書 (たとえば、Key Vault Certificate User または Key Vault Administrator) にアクセスするアクセス許可が必要です。
  • 現在のユーザーに必要なアクセス許可がない場合、Eventstream のこのソースからデータをプレビューすることはできません。

ソースとして Apache Kafka を追加する

まだ eventstream にソースを追加していない場合は、[ データ ソースの接続 ] タイルを選択します。 リボンの [ ソースの追加>データ ソースの接続 ] を選択することもできます。

既に公開されているイベントストリームにソースを追加する場合は、 編集 モードに切り替えます。 リボンで、[ソースの追加] >[データ ソースの接続] を選択します。

[ データ ソースの選択 ] ページまたは [ データ ソース ] ページで、 Apache Kafka を選択します。

Apache Kafka を構成して接続する

1. 接続ページで、新しい接続を選択します。

   

2. [ 接続設定 ] セクションの [ブートストラップ サーバー] に、1 つ以上の Kafka ブートストラップ サーバー アドレスを入力します。 複数のアドレスをコンマ (,) で区切ります。

   

3. [ 接続資格情報 ] セクションで、Apache Kafka クラスターへの既存の接続がある場合は、[ 接続] のドロップダウン リストから選択します。 そうでない場合は、次の手順に従います。

   1. 接続名 には、接続の名前を入力します。
   2. [認証の種類] に [API キー] が選択されていることを確認します。
   3. [キー] と [シークレット] には、API キーとキーのシークレットを入力します。

      注

      mTLS のみを使用して認証を行う場合は、接続の作成時に Key セクションに任意の文字列を追加できます。

4. [接続] を選択します。 

5. 次に、[接続] ページで次の手順に従います。

   1. [トピック] に Kafka トピックを入力します。

   2. [コンシューマー グループ] に Apache Kafka クラスターのコンシューマー グループを入力します。 このフィールドには、イベントを取得するための専用のコンシューマー グループが用意されています。

   3. [自動オフセットのリセット] を選択して、コミットがない場合にオフセットの読み取りを開始する場所を指定します。

   4. [セキュリティ プロトコル] で、次のいずれかのオプションを選択します。

      • SASL_SSL: Kafka クラスターで SASL ベースの認証を使用する場合は、このオプションを使用します。 既定では、Kafka ブローカーのサーバー証明書は、 信頼された CA リストに含まれる証明機関 (CA) によって署名されている必要があります。 Kafka クラスターでカスタム CA が使用されている場合は、 TLS/mTLS 設定を使用して構成できます。
      • SSL (mTLS): Kafka クラスターで mTLS 認証が必要であり、 TLS/mTLS 設定でカスタム サーバー CA 証明書とクライアント証明書の両方を構成する必要がある場合に、このオプションを使用します。

   5. 既定の [SASL メカニズム] は、特に構成されていない限り通常は PLAIN です。 セキュリティ要件に合わせて、SCRAM-SHA-256 または SCRAM-SHA-512 メカニズムを選択できます。

   6. Kafka クラスターでカスタム CA が使用されている場合、または mTLS が必要な場合は、 TLS/mTLS 設定 を展開し、必要に応じて次のオプションを構成します。

      • 信頼 CA 証明書: このオプションを有効にして、サーバー CA 証明書を構成します。 サブスクリプション、リソース グループ、キー コンテナーを選択し、証明書名を指定します。
      • クライアント証明書とキー: このオプションを有効にして、クライアント証明書とキーを構成します。
        • 同じ CA 証明書キー コンテナーを使用する: 両方の証明書が同じキー コンテナーに格納されている場合は、このチェック ボックスをオンにします。 次に、証明書名を指定します。
        • このチェック ボックスをオンにしない場合は、サブスクリプション、リソース グループ、キー コンテナーを選択し、証明書名を指定します。

      注

      プライベート ネットワーク内のソースの場合は、証明書を含むAzure Key Vaultが、Eventstream コネクタの仮想ネットワークインジェクション用にストリーミング仮想ネットワーク データ ゲートウェイによって使用されるAzure仮想ネットワークに接続されていることを確認します (プライベート エンドポイント経由など)。

   

TLS/mTLS 証明書の要件

TLS/mTLS 設定を構成した場合、証明書形式の仕様と、Azure Key Vaultにアップロードするときの一般的な構成ミスについては、このセクションを参照してください。

証明書チェーン

サーバー証明書 SAN の要件

ホスト名検証 () を渡すには、サーバー証明書にブローカーの IP アドレスと DNS 名をサブジェクト代替名 (SAN) に含めるssl.endpoint.identification.algorithm=https。

subjectAltName:
  DNS.1 = {broker FQDN}
  DNS.2 = localhost
  IP.1  = {broker public IP}
  IP.2  = 127.0.0.1

証明書をAzure Key Vaultにアップロードする

証明書は、Azure Key Vault 証明書オブジェクトPEM 形式としてアップロードされます。 PEM バンドル ファイルは、証明書と秘密キーが 1 つのファイルに連結されています。

-----BEGIN CERTIFICATE-----
MIIExjCCA...
-----END CERTIFICATE-----
-----BEGIN RSA PRIVATE KEY-----
MIIEpAIB...
-----END RSA PRIVATE KEY-----

キーのプロパティに一致する インポート ポリシー を使用します。

{
  "secretProperties": {
    "contentType": "application/x-pem-file"
  },
  "keyProperties": {
    "exportable": true,
    "keyType": "RSA",
    "keySize": 4096,
    "reuseKey": false
  },
  "issuerParameters": {
    "name": "Unknown"
  }
}

証明書をインポートするには、次のコマンドを実行します。

az keyvault certificate import \
  --vault-name {kvName} \
  --name {certName} \
  --file {pemBundleFile} \
  --policy @{policyFile}

よくあるミス

更新された Eventstream を表示する

編集モードで、Eventstream に追加された Apache Kafka ソースを確認できます。 

これらの手順を完了すると、Apache Kafka ソースを [ライブ] ビューで視覚化できるようになります。

関連するコンテンツ

その他のコネクタ。

• Amazon Kinesis データ ストリーム
• Azure Cosmos DB
• Azure Event Hubs
• Azure IoT Hub
• Azure SQL Database 変更データ キャプチャ (CDC)
• カスタム エンドポイント
• Google Cloud Pub/Sub
• MySQL データベース CDC
• PostgreSQL データベースの CDC
• サンプル データ
• Azure Blob Storage イベント
• Fabric ワークスペース イベント