[Yandex Cloud documentation](../../../../index.md) > [Yandex Data Transfer](../../../index.md) > [Step-by-step guides](../../index.md) > [Configuring endpoints](../index.md) > Apache Kafka® > Target # Transferring data to a Apache Kafka® target endpoint Yandex Data Transfer enables you to migrate data to a Apache Kafka® queue and implement various data processing and transformation scenarios. To implement a transfer: 1. [Explore possible data transfer scenarios](#scenarios). 1. [Configure one of the supported data sources](#supported-sources). 1. [Configure the target endpoint](#endpoint-settings) in Yandex Data Transfer. 1. [Create](../../transfer.md#create) a transfer and [start](../../transfer.md#activate) it. 1. Perform required operations with the database and [control the transfer](../../monitoring.md). 1. In case of any issues, [use ready-made solutions](../../../troubleshooting/index.md) to resolve them. ## Scenarios for transferring data to Apache Kafka® {#scenarios} 1. Migration: Moving data from one storage to another. Migration often means migrating a database from obsolete local databases to managed cloud ones. Mirroring data across queues is a separate migration task. * [Apache Kafka® mirroring](../../../tutorials/mkf-to-mkf.md) 1. Data change capture means tracking changes to a database and delivering those changes to consumers. It is used for applications that are sensitive to real-time data changes. * [Capturing changes from MySQL® and delivering to Apache Kafka®](../../../tutorials/cdc-mmy.md). * [YDB change data capture and delivery to Apache Kafka®](../../../tutorials/cdc-ydb.md). * [Capturing changes from PostgreSQL and delivering to Apache Kafka®](../../../tutorials/cdc-mpg.md). 1. Data delivery is a process of delivering arbitrary data to target storage. It includes data retrieval from a queue and its deserialization with subsequent transformation to target storage format. * [Delivering data from a YDS queue to Apache Kafka®](../../../tutorials/yds-to-kafka.md) For a detailed description of possible Yandex Data Transfer scenarios, see [Tutorials](../../../tutorials/index.md). ## Configuring the data source {#supported-sources} Configure one of the supported data sources: * [PostgreSQL](../source/postgresql.md) * [MySQL®](../source/mysql.md) * [Apache Kafka®](../source/kafka.md) * [Airbyte®](../../../transfer-matrix.md#airbyte) * [YDS](../source/data-streams.md) * [Managed Service for YDB](../source/ydb.md) * [OpenSearch](../source/opensearch.md) For a complete list of supported sources and targets in Yandex Data Transfer, see [Available transfers](../../../transfer-matrix.md). ## Configuring the Apache Kafka® target endpoint {#endpoint-settings} When [creating](../index.md#create) or [updating](../index.md#update) an endpoint, you can define: * [Yandex Managed Service for Apache Kafka® cluster](#managed-service) connection or [custom installation](#on-premise) settings and [serialization settings](#serializer), including those based on Yandex Compute Cloud VMs. These are required parameters. * [Apache Kafka topic settings](#kafka-settings). ### Managed Service for Apache Kafka® cluster {#managed-service} {% note warning %} To create or edit an endpoint of a managed database, you will need the [`managed-kafka.viewer`](../../../../managed-kafka/security/index.md#mkf-viewer) role or the primitive [`viewer`](../../../../iam/roles-reference.md#viewer) role for the folder the cluster of this managed database resides in. {% endnote %} Connection with the cluster specified in Yandex Cloud. {% list tabs group=instructions %} - Management console {#console} * **Connection type**: Select a cluster connection option: * **Self-managed**: Allows you to specify connection settings manually. Select **Managed Service for Apache Kafka cluster** as the installation type and configure these settings: * **Managed Service for Apache Kafka cluster**: Select the cluster to connect to. * **Authentication**: Select the connection type (`SASL` or `No authentication`). If you select `SASL`: * **Username**: Specify the name of the account, under which Data Transfer will connect to the topic. * **Password**: Enter the account password. * **Connection Manager**: Allows connecting to the cluster via [Yandex Connection Manager](../../../../metadata-hub/quickstart/connection-manager.md): * Select the folder with the Managed Service for Apache Kafka® cluster. * Select **Managed DB cluster** as the installation type and configure these settings: * **Cluster for Managed DB**: Select the cluster to connect to. * **Connection**: Select or create a connection in Connection Manager. {% note warning %} To use a connection from Connection Manager, the user must have [access permissions](../../../../metadata-hub/operations/connection-access.md) for this connection of `connection-manager.user` or higher. {% endnote %} * Under **Topic**, select how you want to specify your topic: * **Topic full name**: Enter the full name of the topic. If you do not want to split the event stream into independent queues by table, enable the **Save transactions order** setting. * **Topic prefix**: Enter the topic prefix. * **Security groups**: Select the cloud network to host the endpoint and security groups for network traffic. Thus, you will be able to apply the specified security group rules to the VMs and clusters in the selected network without changing the settings of these VMs and clusters. For more information, see [Networking in Yandex Data Transfer](../../../concepts/network.md). * Under **Serializing settings**, select the [serialization](../../../concepts/serializer.md) type. For more information, see [Serialization settings](kafka.md#serializer). * Configure the **Advanced settings** of the endpoint, if required. * Specify the configuration of topics in **Configuration name** format: **Configuration value**. * Select compression type. - Terraform {#tf} * Endpoint type: `kafka_target`. * `security_groups`: [Security groups](../../../../vpc/concepts/security-groups.md) for network traffic. Security group rules apply to a transfer. They allow opening network access from the transfer VM to the cluster. For more information, see [Networking in Yandex Data Transfer](../../../concepts/network.md). Security groups must belong to the same network as the cluster. {% note info %} In Terraform, it is not required to specify a network for security groups. {% endnote %} * `connection.cluster_id`: ID of the cluster to connect to. * `auth`: Authentication method used to connect to broker hosts: * `sasl`: SASL authentication. * `no_auth`: Without authentication. Here is an example of the configuration file structure: ```hcl resource "yandex_datatransfer_endpoint" "" { name = "" settings { kafka_target { security_groups = [""] connection { cluster_id = "" } auth { } } } } ``` For more information, see [this Terraform provider guide](../../../../terraform/resources/datatransfer_endpoint.md). - API {#api} * `securityGroups`: Network traffic security groups whose rules apply to VMs and clusters without altering their configurations. For more information, see [Networking in Yandex Data Transfer](../../../concepts/network.md). * `connection.clusterId`: ID of the cluster to connect to. * `auth`: Authentication method used to connect to broker hosts: * `sasl`: SASL authentication. The following parameters are required: * `user`: Name of the account Data Transfer will use to connect to the topic. * `password.raw`: Password for the account in text form. * `mechanism`: Hashing mechanism. * `noAuth`: Without authentication. {% endlist %} ### Custom installation {#on-premise} Connection to the Apache Kafka® cluster with explicitly specified network addresses and broker host ports. {% list tabs group=instructions %} - Management console {#console} * **Connection type**: Select a database connection option: * **Self-managed**: Allows you to specify connection settings manually. Select **Custom installation** as the installation type and configure these settings: * **Broker URLs**: Specify the IP addresses or FQDNs of the broker hosts. If the Apache Kafka® port number differs from the standard one, specify it with a colon after the host name: ```text : ``` * **SSL**: Use encryption to protect the connection. * **PEM Certificate**: If transmitted data has to be be encrypted, e.g., to meet the PCI DSS, upload the [certificate](../../../../managed-kafka/operations/connect/index.md#get-ssl-cert) file or add its contents as text. {% note warning %} If no certificate is added, the transfer may [fail with an error](../../../troubleshooting/index.md#failed-to-connect). {% endnote %} * **Endpoint network interface**: Select or [create](../../../../vpc/operations/subnet-create.md) a subnet in the required [availability zone](../../../../overview/concepts/geo-scope.md). The transfer will use this subnet to access the target. If this field has a value specified for both endpoints, both subnets must be hosted in the same availability zone. * **Authentication**: Select the connection type (`SASL` or `No authentication`). If you select `SASL`: * **Username**: Specify the name of the account, under which Data Transfer will connect to the topic. * **Password**: Enter the account password. * **Mechanism**: Select the hashing mechanism (SHA 256 or SHA 512). * **Connection Manager**: Allows connecting to the database using [Yandex Connection Manager](../../../../metadata-hub/quickstart/connection-manager.md): * Select the folder where the Connection Manager connection was created. * Select **Custom installation** as the installation type and configure these settings: * **Connection**: Select or create a connection in Connection Manager. * **Subnet ID**: Select or [create](../../../../vpc/operations/subnet-create.md) a subnet in the required [availability zone](../../../../overview/concepts/geo-scope.md). The transfer will use this subnet to access the database. If this field has a value specified for both endpoints, both subnets must be hosted in the same availability zone. * Under **Topic**, select how you want to specify your topic: * **Topic full name**: Enter the full name of the topic. If you do not want to split the event stream into independent queues by table, enable the **Save transactions order** setting. * **Topic prefix**: Enter the topic prefix. * **Security groups**: Select the cloud network to host the endpoint and security groups for network traffic. Thus, you will be able to apply the specified security group rules to the VMs and clusters in the selected network without changing the settings of these VMs and clusters. For more information, see [Networking in Yandex Data Transfer](../../../concepts/network.md). * Under **Serializing settings**, select the [serialization](../../../concepts/serializer.md) type. For more information, see [Serialization settings](kafka.md#serializer). * Configure the **Advanced settings** of the endpoint, if required. * Specify the configuration of topics in **Configuration name** format: **Configuration value**. * Select compression type. - Terraform {#tf} * Endpoint type: `kafka_target`. * `security_groups`: [Security groups](../../../../vpc/concepts/security-groups.md) for network traffic. Security group rules apply to a transfer. They allow opening network access from the transfer VM to the broker hosts. For more information, see [Networking in Yandex Data Transfer](../../../concepts/network.md). Security groups must belong to the same network as the `subnet_id` subnet, if the latter is specified. {% note info %} In Terraform, it is not required to specify a network for security groups. {% endnote %} * `connection.on_premise`: Parameters for connecting to the broker hosts: * `broker_urls`: IP addresses or FQDNs of the broker hosts. If the Apache Kafka® port number differs from the standard one, specify it with a colon after the host name: ```text : ``` * `tls_mode`: Parameters for encrypting the data to transfer, if required, e.g., for compliance with the PCI DSS requirements. * `disabled`: Disabled. * `enabled`: Enabled. * `ca_certificate`: CA certificate. {% note warning %} If no certificate is added, the transfer may [fail with an error](../../../troubleshooting/index.md#failed-to-connect). {% endnote %} * `subnet_id`: ID of the subnet hosting the broker hosts. The transfer will use this subnet to access them. * `auth`: Authentication method used to connect to broker hosts: * `sasl`: SASL authentication. * `no_auth`: Without authentication. Here is an example of the configuration file structure: ```hcl resource "yandex_datatransfer_endpoint" "" { name = "" settings { kafka_target { security_groups = [""] connection { on_premise { broker_urls = [""] subnet_id = "" } } auth = { } } } } ``` For more information, see [this Terraform provider guide](../../../../terraform/resources/datatransfer_endpoint.md). - API {#api} * `securityGroups`: Network traffic security groups whose rules apply to VMs and clusters without altering their configurations. For more information, see [Networking in Yandex Data Transfer](../../../concepts/network.md). * `connection.onPremise`: Parameters for connecting to the broker hosts: * `brokerUrls`: IP addresses or FQDNs of the broker hosts. If the Apache Kafka® port number differs from the standard one, specify it with a colon after the host name: ```text : ``` * `tlsMode`: Parameters for encrypting the data to transfer, if required, e.g., for compliance with the PCI DSS requirements. * `disabled`: Disabled. * `enabled`: Enabled. * `caCertificate`: CA certificate. {% note warning %} If no certificate is added, the transfer may [fail with an error](../../../troubleshooting/index.md#failed-to-connect). {% endnote %} * `subnetId`: ID of the subnet hosting the broker hosts. The transfer will use this subnet to access them. * `auth`: Authentication method used to connect to broker hosts: * `sasl`: SASL authentication. The following parameters are required: * `user`: Name of the account Data Transfer will use to connect to the topic. * `password.raw`: Password for the account in text form. * `mechanism`: Hashing mechanism. * `noAuth`: Without authentication. {% endlist %} ### Apache Kafka® topic settings {#kafka-settings} {% list tabs group=instructions %} - Management console {#console} * **Topic**: * **Topic full name**: Specify the name of the topic to send messages to. Select **Save transactions order**, so as not to split an event stream into independent queues by table. * **Topic prefix**: Specify the topic prefix, similar to the `Debezium database.server.name` setting. Messages will go to the topic named `..`. - Terraform {#tf} Under `topic_settings`, specify one of the options to send messages to a topic: * `topic`: Specify parameters in this section to send all messages to one topic: * `topic_name`: Name of the topic to send messages to. * `save_tx_order`: Option for saving the order of transactions. Set to `true` not to split the event stream into independent queues by table. * `topic_prefix`: Specify a prefix to send messages to different topics with the given prefix. It is similar to the Debezium `database.server.name` setting. Messages will be sent to a topic named `..`. - API {#api} In the `topicSettings` field, specify one of the options to send messages to a topic: * `topic`: Specify parameters in this field to send all messages to one topic: * `topicName`: Name of the topic to send messages to. * `saveTxOrder`: Option for saving the order of transactions. Set to `true` not to split the event stream into independent queues by table. * `topicPrefix`: Specify a prefix to send messages to different topics with the given prefix. It is similar to the Debezium `database.server.name` setting. Messages will be sent to a topic named `..`. {% endlist %} Yandex Data Transfer supports CDC for transfers from PostgreSQL, MySQL®, and YDB databases to Apache Kafka® and Yandex Data Streams. Data is sent to the target in Debezium format. For more information about the CDC mode, see [Change data capture](../../../concepts/cdc.md). {% note info %} In YDB, CDC mode is supported starting from version 22.5. {% endnote %} ### Serialization settings {#serializer} {% list tabs group=instructions %} - Management console {#console} * Under **Serializing settings**, select the [serialization](../../../concepts/serializer.md) type: * **Auto**: Automatic serialization. * **Debezium**: Serialization under the Debezium standards: * **Yandex Schema Registry**: * Select a message schema (data format). * From the list, select the [namespace](../../../../metadata-hub/operations/list-name-space.md) of the Yandex Schema Registry schema registry. * **On Premise**: * Select the message key schema (matches the `key.converter` Debezium parameter). The default value matches the `key.converter.schemas.enable=true` and `key.converter=org.apache.kafka.connect.json.JsonConverter` Debezium parameters. If you select **Disabled**, the `key.converter.schemas.enable` Debezium parameter takes the `false` value. * Select the message value schema (matches the `value.converter` Debezium parameter). The default value matches the `value.converter.schemas.enable=true` and `value.converter=org.apache.kafka.connect.json.JsonConverter` Debezium parameters. If you select **Disabled**, the `value.converter.schemas.enable` Debezium parameter takes the `false` value. * When selecting the **Avro (via Schema Registry)**, **JSON (via Schema Registry)**, or **Protobuf (via Schema Registry)** value, specify the parameters for access to Confluent Schema Registry for message key and value schemas: * **URL**: URL. * **Username**. This is an optional setting. * **Password**. This is an optional setting. * **CA certificate**: Upload the SSL certificate for the connection. This is an optional setting. To skip the certificate check, do not provide one. * If required, specify [**Debezium serializer settings**](../../../concepts/serializer.md#debezium) in `Parameter`-`Value` format. If you want to use JSON schemas in Yandex Schema Registry and preserve their [compatibility when adding and deleting optional fields](../../../../metadata-hub/concepts/schema-registry-content-model.md#optional-parameters-compatibility-solution), use these settings: **Serializing settings**: **Debezium**: **On Premise**. * To use Schema Registry for keys, select **Key schema type**: **JSON (via Schema Registry)**. To use Schema Registry for values, select **Value schema type**: **JSON (via Schema Registry)**. Set the parameters for access to Schema Registry: * **URL**: `https://.schema-registry.yandexcloud.net:443`. {% note warning %} The namespace must have **Compatibility check policy for JSON** [set](../../../../metadata-hub/operations/update-name-space.md) to `optional friendly`. {% endnote %} * **Username**: `api-key`. * **Password**: Value of the [API key](../../../../iam/concepts/authorization/api-key.md) with a limited scope used for connecting to Schema Registry. To get this value: 1. Create an API key with a limited scope and place it in the `SECRET` local variable: ```bash yc iam api-key create --folder-id \ --service-account-name \ --scopes yc.schema-registry.schemas.manage \ --expires-at '2030-01-01T00:00:00Z' >./api-key.yaml && \ SECRET=`cat ./api-key.yaml | grep 'secret:' | awk '{print $2}'` ``` 1. Print the `SECRET` variable value to the terminal: ```bash echo $SECRET ``` * Under **Debezium serializer settings**: * To generate a closed schema for keys, include the `key.converter.dt.json.generate.closed.content.schema` parameter set to `true`. * To generate a closed schema for values, include the `value.converter.dt.json.generate.closed.content.schema` parameter set to `true`. - Terraform {#tf} Under `serializer`, specify the selected [serialization](../../../concepts/serializer.md) type: * `serializer_auto` * `serializer_json` * `serializer_debezium` For this type, you can specify the [Debezium serialization](../../../concepts/serializer.md#debezium) parameters as `key`/`value` pairs in the `serializer_debezium.serializer_parameters` field. If you want to use JSON schemas in Yandex Schema Registry and preserve their [compatibility when adding and deleting optional fields](../../../../metadata-hub/concepts/schema-registry-content-model.md#optional-parameters-compatibility-solution), add the `serializer` section with a description of the serialization settings to the configuration file. To generate a closed schema for keys, include the `key.converter.dt.json.generate.closed.content.schema` variable set to `true` in the `serializer` section. To generate a closed schema for values, include the `value.converter.dt.json.generate.closed.content.schema` variable set to `true` in the `serializer` section. ```hcl resource "yandex_datatransfer_endpoint" "" { ... settings { kafka_target { ... serializer { serializer_debezium { serializer_parameters { key = "key.converter.dt.json.generate.closed.content.schema" value = "true" } serializer_parameters { key = "value.converter.dt.json.generate.closed.content.schema" value = "true" } serializer_parameters { key = "value.converter.schemas.enable" value = "true" } serializer_parameters { key = "value.converter.schema.registry.url" value = "" } serializer_parameters { key = "value.converter.basic.auth.user.info" value = "api-key:" } } } } } } ``` Where: * `Schema_Registry_namespace_URL`: Schema Registry namespace endpoint. You can copy the endpoint from the details for the Schema Registry namespace connection on the **Debezium** tab, in the `value.converter.schema.registry.url` parameter. {% note warning %} The namespace must have **Compatibility check policy for JSON** [set](../../../../metadata-hub/operations/update-name-space.md) to `optional friendly`. {% endnote %} * `API_key_value`: Value of the [API key](../../../../iam/concepts/authorization/api-key.md) with a limited scope used for connecting to Schema Registry. To get this value: 1. Create an API key with a limited scope and place it in the `SECRET` local variable: ```bash yc iam api-key create --folder-id \ --service-account-name \ --scopes yc.schema-registry.schemas.manage \ --expires-at '2030-01-01T00:00:00Z' >./api-key.yaml && \ SECRET=`cat ./api-key.yaml | grep 'secret:' | awk '{print $2}'` ``` 1. Print the `SECRET` variable value to the terminal: ```bash echo $SECRET ``` 1. Copy the printed value and paste it into the `value` parameter in the configuration file. - API {#api} Under `serializer`, specify the selected [serialization](../../../concepts/serializer.md) type: * `serializerAuto` * `serializerJson` * `serializerDebezium` For this type, you can specify the [Debezium serialization](../../../concepts/serializer.md#debezium) parameters as `key`/`value` pairs in the `serializerDebezium.serializerParameters` field. If you want to use JSON schemas in Yandex Schema Registry and preserve their [compatibility when adding and deleting optional fields](../../../../metadata-hub/concepts/schema-registry-content-model.md#optional-parameters-compatibility-solution), add the `serializer` object with a description of the serialization settings to the request body. To generate a closed schema for keys, add the `key.converter.dt.json.generate.closed.content.schema` variable set to `true` in `serializer`. To generate a closed schema for values, add the `value.converter.dt.json.generate.closed.content.schema` variable set to `true` in `serializer`. ```json "serializer": { "serializerDebezium": { "serializerParameters": [ { "key": "converter.dt.json.generate.closed.content.schema", "value": "true" }, { "key": "value.converter", "value": "io.confluent.connect.json.JsonSchemaConverter" }, { "key": "value.converter.schemas.enable", "value": "true" }, { "key": "value.converter.schema.registry.url", "value": "" }, { "key": "value.converter.basic.auth.user.info", "value": "api-key:" } ] } } ``` Where: * `Schema_Registry_namespace_URL`: Schema Registry namespace endpoint. You can copy the endpoint from the details for the Schema Registry namespace connection on the **Debezium** tab, in the `value.converter.schema.registry.url` parameter. {% note warning %} The namespace must have **Compatibility check policy for JSON** [set](../../../../metadata-hub/operations/update-name-space.md) to `optional friendly`. {% endnote %} * `API_key_value`: Value of the [API key](../../../../iam/concepts/authorization/api-key.md) with a limited scope used for connecting to Schema Registry. To get this value: 1. Create an API key with a limited scope and place it in the `SECRET` local variable: ```bash yc iam api-key create --folder-id \ --service-account-name \ --scopes yc.schema-registry.schemas.manage \ --expires-at '2030-01-01T00:00:00Z' >./api-key.yaml && \ SECRET=`cat ./api-key.yaml | grep 'secret:' | awk '{print $2}'` ``` 1. Print the `SECRET` variable value to the terminal: ```bash echo $SECRET ``` 1. Copy the printed value and paste it into the `value` parameter in the configuration file. {% endlist %} ### Additional settings {#additional-settings} {% list tabs group=instructions %} - Management console {#console} You can specify [topic configuration parameters](https://docs.confluent.io/platform/current/installation/configuration/topic-configs.html) to use when creating new topics. Specify the parameter and one of its possible values, e.g., `cleanup.policy` and `compact`. {% endlist %} After configuring the data source and target, [create and start the transfer](../../transfer.md#create).