Azure Cognitive Search Sink Connector for Confluent Cloud¶

注釈

If you are installing the connector locally for Confluent Platform, see Azure Cognitive Search Sink Connector for Confluent Platform.

Azure Cognitive Search Sink Connector for Confluent Cloud を使用すると、Apache Kafka® から Azure Cognitive Search にデータを移動できます。このコネクターは、各イベントを Kafka のトピックから（ドキュメントとして）|az| Cognitive Search のインデックスに書き込みます。コネクターでは、レコードをドキュメントとして送信するために、Azure Cognitive Search の REST API が使用されます。

機能¶

Azure Cognitive Search Sink Connector では、以下の機能をサポートしています。

少なくとも 1 回のデリバリー: コネクターによって、Kafka のトピックからのレコードが少なくとも 1 回は配信されることが保証されます。
複数のタスクのサポート: このコネクターは、1 つまたは複数のタスクの実行をサポートしています。タスクが多いほどパフォーマンスが向上する可能性があります。
順序どおりの書き込み: レコードはコネクターによって受信されたとおりの順序で書き込まれます。また、一意性確保のために、ドキュメントキーとして Kafka 座標（トピック、パーティション、オフセット）を使用できます。それ以外の場合は、コネクターでドキュメントキーとしてレコードキーが使用されます。
トピックの自動作成: コネクターの起動時に、以下の 3 つのトピックが自動的に作成されます。
- 成功トピック
- エラートピック
- デッドレターキュー（DLQ）トピック
各トピック名にはサフィックスとして、コネクターの論理 ID が付けられます。以下の例では、コネクターのトピックが 3 つと、pageviews という名前の既存の Kafka トピックが 1 つ含まれています。

コネクターのトピック¶

トピックに送信されたレコードが正しいフォーマットではない場合、またはレコード内に重要なフィールドが存在しない場合は、エラートピックにエラーが記録され、コネクターは動作を継続します。
自動再試行: Azure Cognitive Search サービスで障害が発生した場合は、すべてのリクエスト（再試行可能なもの）がコネクターによって再試行されます。コネクターによる再試行期間の最大値を指定するには、max.retry.ms 構成プロパティを使用します。
サポートされるデータフォーマット: このコネクターは、Avro、JSON スキーマ（JSON-SR）、および Protobuf 入力フォーマットをサポートします。これらのスキーマレジストリベースのフォーマットを使用するには、Schema Registry を有効にしておく必要があります。

See Cloud connector limitations for more information.

Azure サービスプリンシパル¶

コネクターを実行するには、Azure RBAC サービスプリンシパルが必要です。サービスプリンシパルを作成する場合の Azure CLI コマンドからの出力結果に、必要な認証および認可の詳細が含まれます。これをコネクターの構成に追加します。

注釈

CLI の代わりに Azure ポータルを使用して既存のサービスプリンシパルにロールを割り当てる場合は、「Azure portal を使用して Azure ロールを割り当てる」を参照してください。

Azure CLI を使用してサービスプリンシパルキーを作成するには、以下の手順を実行します。

Azure CLI にログインします。
```
az login
```

以下のコマンドを入力して、サービスプリンシパルを作成します。

az ad sp create-for-rbac --name <Name of service principal> --scopes \
/subscriptions/<SubscriptionID>/resourceGroups/<Resource_Group>

以下に例を示します。

az ad sp create-for-rbac --name azure_search --scopes /subscriptions/
d92eeba4-...omitted...-37c2bd9259d0/resourceGroups/connect-azure

Creating 'Contributor' role assignment under scope '/subscriptions/
d92eeba4-...omitted...-37c2bd9259d0/resourceGroups/connect-azure'

The output includes credentials that you must protect. Be sure that you
do not include these credentials in your code or check the credentials
into your source control.

{
   "appId": "8ec186f9-...omitted...-e575b928b00a",
   "displayName": "azure_search",
   "name": "8ec186f9-...omitted...-e575b928b00a",
   "password": "jdGzGTwCKQ...omitted...QwE3hx",
   "tenant": "0893715b-...omitted...-2789e1ead045"
}

コネクター構成で使用するために、以下の詳細を保存します。

"appId" の出力は、コネクターの UI フィールド Azure Client ID （CLI プロパティ azure.search.client.id）に使用します。
"password" の出力は、コネクターの UI フィールド Azure Client Secret （CLI プロパティ azure.search.client.secret）に使用します。
"tenant" の出力は、コネクターの UI フィールド Azure Tenant ID （CLI プロパティ azure.search.tenant.id）に使用します。
ちなみに

必要な場合は、サービスプリンシパルをさらに細かく設定できます。たとえば、以下のコマンドでは、特に Azure 検索サービスにアクセスするための contributor ロールの割り当てを作成します。
```
az ad sp create-for-rbac --name <Name of service principal> --scopes
 /subscriptions/<SubscriptionID>/resourceGroups/<Resource Group>
 /providers/Microsoft.Search/searchServices/<Search Service Name>
 --role Reader
```

クイックスタート¶

このクイックスタートを使用して、Confluent Cloud Azure Cognitive Search Sink Connector の利用を開始することができます。このクイックスタートでは、コネクターを選択し、イベントをストリーミングするようにコネクターを構成するための基本的な方法について説明します。

前提条件

Microsoft Azure （Azure）上の Confluent Cloud クラスターへのアクセスを許可されていること。
Azure サービスプリンシパル、Azure Cognitive Search の API キー、および契約プランの詳細情報（コネクター構成で使用）。
Confluent CLI がインストールされ、クラスター用に構成されていること。「Confluent CLI のインストール」を参照してください。
スキーマレジストリベースのフォーマット（Avro、JSON_SR（JSON スキーマ）、Protobuf など）を使用するには、Schema Registry を有効にしておく必要があります。詳細については、「環境の制限」を参照してください。
Azure Cognitive Search に少なくとも 1 つのインデックスが存在する必要があります。
すべてのレコードスキーマフィールドが Azure 検索サービスのインデックスフィールドとして存在する必要があります。
シンクコネクターを作成する前に、Confluent Cloud クラスター上にソース Kafka トピックが 1 つ以上存在している必要があります。

Confluent Cloud Console の使用¶

ステップ 1: Confluent Cloud クラスターを起動します。¶

インストール手順については、「Confluent Cloud を使用した Apache Kafka のクイックスタート」を参照してください。

ステップ 2: コネクターを追加します。¶

左のナビゲーションメニューの Data integration をクリックし、Connectors をクリックします。クラスター内に既にコネクターがある場合は、+ Add connector をクリックします。

ステップ 3: コネクターを選択します。¶

Azure Cognitive Search Sink コネクターのアイコンをクリックします。

Step 4: Enter the connector details.¶

注釈

Ensure you have all your prerequisites completed.
アスタリスク（ * ）は必須項目であることを示しています。

At the Add Azure Cognitive Search Sink Connector screen, complete the following:

If you've already populated your Kafka topics, select the topic(s) you want to connect from the Topics list.

To create a new topic, click +Add new topic.

Enter your Solace connection details:
- Azure Search Service Name: The name of the Azure Search service.
- Azure Search API Key: The API key for the Azure Search service.
- Azure Client ID: Client ID of service principal of your subscription.
- Azure Client Secret: Client secret of service principal of your subscription.
- Azure Tenant ID: Tenant ID of service principal of your subscription.
- Azure Subscription ID: Azure subscription ID for your Azure account.
- ResourceGroup Name: ResourceGroup in which Azure Search service exists.
Click Continue.

注釈

Configuration properties that are not shown in the Cloud Console use the default values. See 構成プロパティ for all property values and descriptions.

Select the Input Kafka record value format (data coming from the Kafka topic): AVRO, JSON_SR (JSON Schema), or PROTOBUF. A valid schema must be available in Schema Registry to use a schema-based message format (for example, AVRO, JSON_SR, or PROTOBUF.
Enter the Index Pattern Name, which is the name of the index to write records as documents to. Use ${topic} within the pattern to specify the topic of the record.
Show advanced configurations
- Write Method: The method used to write Kafka records to an index. Available methods are Upload which functions like upsert and MergeOrUpload, which updates an existing document with the specified fields. If the document doesn't exist, it behaves like Upload.
- Delete Enabled: Whether documents will be deleted if the record value is null.
- Key Mode: Determines what will be used for the document key id. The available modes are:
  - KEY: The Kafka record key is used as the document key.
  - COORDINATES: The Kafka coordinates (topic, partition, and offset) are concatenated to form the document key. This allows for unique document keys.
- Max Batch Size: The maximum number of Kafka records that will be sent per request. To disable batching of records, set this value to 1.
- Maximum Retry Time (ms): The maximum amount of time in milliseconds that the connector will attempt its request before aborting it.
For information about transforms and predicates, see the Single Message Transforms (SMT) documentation for details. See サポートされない変換 for a list of SMTs that are not supported with this connector.
Click Continue.

Step 5: Check for documents.¶

ドキュメントが検索インデックスに取り込まれていることを確認します。

Connect 用の Confluent Cloud API の使用に関する詳細とサンプルについては、「Confluent Cloud API for Connect」セクションを参照してください。

ちなみに

コネクターを起動すると、デッドレターキューのトピックが自動的に作成されます。詳細については、「Confluent Cloud デッドレターキュー」を参照してください。

Confluent CLI の使用¶

以下の手順に従うと、Confluent CLI を使用してコネクターをセットアップし、実行できます。

注釈

すべての前提条件を満たしていることを確認してください。
コマンド例では Confluent CLI バージョン 2 を使用しています。詳細については、「Confluent CLI v2 への移行 <https://docs.confluent.io/confluent-cli/current/migrate.html#cli-migrate>`__」を参照してください。

ステップ 1: 使用可能なコネクターをリスト表示します。¶

以下のコマンドを入力して、使用可能なコネクターをリスト表示します。

confluent connect plugin list

ステップ 2: コネクターの必須の構成プロパティを表示します。¶

以下のコマンドを入力して、コネクターの必須プロパティを表示します。

confluent connect plugin describe <connector-catalog-name>

以下に例を示します。

confluent connect plugin describe AzureCognitiveSearchSink

出力例:

Following are the required configs:
connector.class: AzureCognitiveSearchSink
input.data.format
name
kafka.api.key
kafka.api.secret
azure.search.service.name
azure.search.api.key
azure.search.client.id
azure.search.client.secret
azure.search.tenant.id
azure.search.subscription.id
azure.search.resourcegroup.name
index.name
tasks.max
topics

ステップ 3: コネクターの構成ファイルを作成します。¶

コネクター構成プロパティを含む JSON ファイルを作成します。以下の例は、コネクターの必須プロパティを示しています。

{
  "connector.class": "AzureCognitiveSearchSink",
  "input.data.format": "AVRO",
  "name": "AzureCognitiveSearchSink_0",
  "kafka.api.key": "****************",
  "kafka.api.secret": "************************************************",
  "azure.search.service.name": "<service_name>",
  "azure.search.api.key": "<api_key>",
  "azure.search.client.id": "<client_id>",
  "azure.search.client.secret": "<client_secret>",
  "azure.search.tenant.id": "<tenant_id>",
  "azure.search.subscription.id": "<subscription_id>",
  "azure.search.resourcegroup.name": "<resource_group>",
  "index.name": "<index_name>",
  "tasks.max": "1",
  "topics": "<topic_name>"
}

以下のプロパティ定義に注意してください。

"connector.class": コネクターのプラグイン名を指定します。
"input.data.format": Kafka 入力レコード値のフォーマット（Kafka トピックから送られるデータ）を設定します。指定可能なエントリは、AVRO、JSON_SR、および PROTOBUF です。スキーマベースのメッセージフォーマット（たとえば、Avro、JSON_SR（JSON スキーマ）、および Protobuf）を使用するには、Confluent Cloud Schema Registry を構成しておく必要があります。
"name": 新しいコネクターの名前を設定します。
"kafka.api.key" および "kafka.api.secret": これらの認証情報として、クラスター API キーとシークレットを使用するか、サービスアカウントの API キーとシークレットを使用します。
azure.search.<...>: 必須の Azure および Azure 検索接続の詳細情報。プロパティの詳細については、「Azure サービスプリンシパル」および Azure Cognitive Search の API キーを参照してください。
"index.name": レコードを（ドキュメントとして）書き込む先の検索インデックスの名前。
"tasks.max": このコネクターで使用できるタスクの最大数を入力します。タスクが多いほどパフォーマンスが向上する可能性があります。
"topics": 特定のトピック名を指定するか、複数のトピック名をコンマ区切りにしたリストを指定します。

Single Message Transforms: CLI を使用した SMT の追加の詳細については、Single Message Transforms（SMT）のドキュメントを参照してください。このコネクターでサポートされていない SMT のリストについては、「サポートされない変換」を参照してください。

See 構成プロパティ for all property values and descriptions.

ステップ 4: プロパティファイルを読み込み、コネクターを作成します。¶

以下のコマンドを入力して、構成を読み込み、コネクターを起動します。

confluent connect create --config <file-name>.json

以下に例を示します。

confluent connect create --config azure-search-sink-config.json

出力例:

Created connector AzureCognitiveSearchSink_0 lcc-do6vzd

ステップ 5: コネクターのステータスを確認します。¶

以下のコマンドを入力して、コネクターのステータスを確認します。

confluent connect list

出力例:

ID           |             Name             | Status  | Type | Trace
+------------+------------------------------+---------+------+-------+
lcc-do6vzd   | AzureCognitiveSearchSink_0   | RUNNING | sink |       |

ステップ 6: ドキュメントを確認します。¶

Azure 検索インデックスに取り込まれていることを確認します。

Connect 用の Confluent Cloud API の使用に関する詳細とサンプルについては、「Confluent Cloud API for Connect」セクションを参照してください。

ちなみに

コネクターを起動すると、デッドレターキューのトピックが自動的に作成されます。詳細については、「Confluent Cloud デッドレターキュー」を参照してください。

構成プロパティ¶

Use the following configuration properties with this connector.

Which topics do you want to get data from?¶

topics

Identifies the topic name or a comma-separated list of topic names.

Type: list
重要度: 高

Input messages¶

input.data.format

Sets the input Kafka record value format. Valid entries are AVRO, JSON_SR and PROTOBUF. Note that you need to have Confluent Cloud Schema Registry configured

型: string
重要度: 高

How should we connect to your data?¶

name

Sets a name for your connector.

型: string
Valid Values: A string at most 64 characters long
重要度: 高

Kafka Cluster credentials¶

kafka.auth.mode

Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode.

型: string
Default: KAFKA_API_KEY
Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY
重要度: 高

kafka.api.key

型: password
重要度: 高

kafka.service.account.id

The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.

型: string
重要度: 高

kafka.api.secret

型: password
重要度: 高

How should we connect to your Azure Search Service¶

azure.search.service.name

The name of the Azure Search service

型: string
重要度: 高

azure.search.api.key

The api key for the Azure Search service

型: password
重要度: 高

azure.search.client.id

Client ID of service principal of your subscription

型: password
重要度: 高

azure.search.client.secret

Client Secret of service principal of your subscription

型: password
重要度: 高

azure.search.tenant.id

Tenant ID of service principal of your subscription

型: password
重要度: 高

azure.search.subscription.id

Azure Subscription ID for your Azure Account

型: password
重要度: 高

azure.search.resourcegroup.name

ResourceGroup in which Azure Search Service exists

型: string
重要度: 高

Search Service Write Details¶

index.name

The name of the index to write records as documents to. Use ${topic} within the pattern to specify the topic of the record

型: string
重要度: 高

write.method

The method used to write Kafka records to an index. Available methods are Upload - Functions like upsert. A document is inserted if it does not existed and updated/replaced if it does MergeOrUpload - Updates an existing document with the specified fields. If the document doesn't exist, behaves like Upload

型: string
Default: Upload
重要度: 高

delete.enabled

Whether documents will be deleted if the record value is null

Type: boolean
Default: false
重要度: 高

key.mode

Determines what will be used for the document key id. The available modes are:KEY - the Kafka record key is used as the document key COORDINATES - the Kafka coordinates (topic, partition, and offset) are concatenated to form the document key. This allows for unique document keys

型: string
Default: KEY
重要度: 中

max.batch.size

The maximum number of Kafka records that will be sent per request. To disable batching of records, set this value to 1

型: int
Default: 1
Valid Values: [1,...,1000]
重要度: 高

max.retry.ms

The maximum amount of time in ms that the connector will attempt its request before aborting it

型: int
Default: 300000 (5 minutes)
Valid Values: [0,...]
Importance: low

Number of tasks for this connector¶

tasks.max

型: int
Valid Values: [1,...]
重要度: 高

次のステップ¶

参考

フルマネージド型の Confluent Cloud コネクターが Confluent Cloud ksqlDB でどのように動作するかを示す例については、「Cloud ETL のデモ」を参照してください。この例では、Confluent CLI を使用して Confluent Cloud のリソースを管理する方法についても説明しています。