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 をクリックします。
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.
- Select the way you want to provide Kafka Cluster credentials. You can
choose one of the following options:
- Global Access: Allows your connector to access everything you have access to. With global access, connector access will be linked to your account. This option is not recommended for production.
- Granular access: Limits the access for your connector. You will be able to manage connector access through a service account. This option is recommended for production.
- Use an existing API key: Allows you to enter an API key and secret part you have stored. You can enter an API key and secret (or generate these in the Cloud Console).
- Click Continue.
- 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 likeupsert
andMergeOrUpload
, which updates an existing document with the specified fields. If the document doesn't exist, it behaves likeUpload
.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.
Based on the number of topic partitions you select, you will be provided with a recommended number of tasks.
- To change the number of recommended tasks, enter the number of tasks for the connector to use in the Tasks field.
- 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>`__」を参照してください。
ステップ 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 doesMergeOrUpload
- Updates an existing document with the specified fields. If the document doesn't exist, behaves likeUpload
- 型: 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 keyCOORDINATES
- 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 のリソースを管理する方法についても説明しています。