[Документация Yandex Cloud](../../../../index.md) > [Yandex Data Transfer](../../../index.md) > [Пошаговые инструкции](../../index.md) > [Настройка эндпоинтов](../index.md) > Apache Kafka® > Приемник # Передача данных в эндпоинт-приемник Apache Kafka® С помощью сервиса Yandex Data Transfer вы можете переносить данные в очередь Apache Kafka® и реализовывать различные сценарии обработки и трансформации данных. Для реализации трансфера: 1. [Ознакомьтесь с возможными сценариями передачи данных](#scenarios). 1. [Настройте один из поддерживаемых источников данных](#supported-sources). 1. [Настройте эндпоинт-приемник](#endpoint-settings) в Yandex Data Transfer. 1. [Создайте](../../transfer.md#create) и [запустите](../../transfer.md#activate) трансфер. 1. Выполняйте необходимые действия по работе с базой и [контролируйте трансфер](../../monitoring.md). 1. При возникновении проблем, [воспользуйтесь готовыми решениями](../../../troubleshooting/index.md) по их устранению. ## Сценарии передачи данных в Apache Kafka® {#scenarios} 1. Миграция — перенос данных из одного хранилища в другое. Часто это перенос базы из устаревших локальных баз в управляемые облачные. Отдельной задачей миграции является зеркалирование данных между очередями: * [Зеркалирование Apache Kafka®](../../../tutorials/mkf-to-mkf.md) 1. Захват изменений данных — это процесс отслеживания изменений в базе данных и поставка этих изменений потребителям. Применяется для приложений, которые чувствительны к изменению данных в реальном времени. * [Захват изменений из MySQL® и поставка в Apache Kafka®](../../../tutorials/cdc-mmy.md); * [Захват изменений из PostgreSQL и поставка в Apache Kafka®](../../../tutorials/cdc-mpg.md). 1. Поставка данных — процесс доставки произвольных данных в целевые хранилища. Процесс поставки включает извлечение данных из очереди и их десериализацию с последующей трансформацией данных в формат целевого хранилища. * [Поставка данных из очереди YDS в Apache Kafka®](../../../tutorials/yds-to-kafka.md) Подробное описание возможных сценариев передачи данных в Yandex Data Transfer читайте в разделе [Практические руководства](../../../tutorials/index.md). ## Настройка источника данных {#supported-sources} Настройте один из поддерживаемых источников данных: * [PostgreSQL](../source/postgresql.md) * [MySQL®](../source/mysql.md) * [Apache Kafka®](../source/kafka.md) * [Airbyte®](../../../transfer-matrix.md#airbyte) * [YDS](../source/data-streams.md) * [OpenSearch](../source/opensearch.md) Полный список поддерживаемых источников и приемников в Yandex Data Transfer читайте в разделе [Доступные трансферы](../../../transfer-matrix.md). ## Настройка эндпоинта-приемника Apache Kafka® {#endpoint-settings} При [создании](../index.md#create) или [изменении](../index.md#update) эндпоинта вы можете задать: * Настройки подключения к [кластеру Yandex Managed Service for Apache Kafka®](#managed-service) или [пользовательской инсталляции](#on-premise) и [настройки сериализации](#serializer), в т. ч. на базе виртуальных машин Yandex Compute Cloud. Это обязательные параметры. * [Настройки топика Apache Kafka](#kafka-settings). ### Кластер Managed Service for Apache Kafka® {#managed-service} {% note warning %} Для создания или редактирования эндпоинта управляемой базы данных вам потребуется [роль `managed-kafka.viewer`](../../../../managed-kafka/security/index.md#mkf-viewer) или примитивная [роль `viewer`](../../../../iam/roles-reference.md#viewer), выданная на каталог кластера этой управляемой базы данных. {% endnote %} Подключение с указанием кластера в Yandex Cloud. {% list tabs group=instructions %} - Консоль управления {#console} * **Тип подключения** — выберите вариант подключения к кластеру: * **Ручная настройка** — позволяет задать настройки подключения вручную. Выберите тип инсталляции **Кластер Managed Service for Apache Kafka** и задайте настройки: * **Кластер Managed Service for Apache Kafka** — выберите кластер, к которому необходимо подключиться. * **Аутентификация** — выберите тип: `SASL` или `Без аутентификации`. При выборе значения `SASL`: * **Имя пользователя** — укажите имя учетной записи, от имени которой сервис Data Transfer будет подключаться к топику. * **Пароль** — укажите пароль учетной записи. * **Connection Manager** — позволяет подключиться к кластеру через [Yandex Connection Manager](../../../../metadata-hub/quickstart/connection-manager.md): * Выберите каталог, в котором находится кластер Managed Service for Apache Kafka®. * Выберите тип инсталляции **Кластер управляемой БД** и задайте настройки: * **Кластер управляемой БД** — выберите кластер, к которому необходимо подключиться. * **Подключение** — выберите подключение в Connection Manager или создайте его. {% note warning %} Чтобы использовать подключение из Connection Manager, у пользователя должны быть [права доступа](../../../../metadata-hub/operations/connection-access.md) не ниже `connection-manager.user` к этому подключению. {% endnote %} * В блоке **Топик** выберите способ указания топика: * **Полное имя топика** — введите полное имя топика. Если вы не хотите разбивать поток событий на независимые очереди по таблицам, включите настройку **Сохранять порядок транзакций**. * **Префикс топика** — введите префикс топика. * **Группы безопасности** — выберите облачную сеть для размещения эндпоинта и группы безопасности для сетевого трафика. Это позволит применить к ВМ и кластерам в выбранной сети указанные правила групп безопасности без изменения настроек этих ВМ и кластеров. Подробнее читайте в разделе [Сеть в Yandex Data Transfer](../../../concepts/network.md). * В блоке **Настройки сериализации** выберите тип [сериализации](../../../concepts/serializer.md). Подробнее читайте в разделе [Настройки сериализации](kafka.md#serializer). * При необходимости задайте **Дополнительные настройки** эндпоинта: * Укажите конфигурацию топиков в формате **Имя конфигурации**-**Значение конфигурации**. * Выберите тип сжатия. - Terraform {#tf} * Тип эндпоинта — `kafka_target`. * `security_groups` — [группы безопасности](../../../../vpc/concepts/security-groups.md) для сетевого трафика. Правила групп безопасности применяются к трансферу. Они позволяют открыть сетевой доступ с ВМ трансфера к кластеру. Подробнее читайте в разделе [Сеть в Yandex Data Transfer](../../../concepts/network.md). Группы безопасности должны принадлежать той же сети, в которой размещен кластер. {% note info %} В Terraform сеть для групп безопасности задавать не нужно. {% endnote %} * `connection.cluster_id` — идентификатор кластера, к которому необходимо подключиться. * `auth` — метод аутентификации. Используется при подключении к хостам-брокерам: * `sasl` — аутентификация с помощью SASL. * `no_auth` — без аутентификации. Пример структуры конфигурационного файла: ```hcl resource "yandex_datatransfer_endpoint" "<имя_эндпоинта_в_Terraform>" { name = "<имя_эндпоинта>" settings { kafka_target { security_groups = ["<список_идентификаторов_групп_безопасности>"] connection { cluster_id = "<идентификатор_кластера>" } auth { <метод_аутентификации> } <настройки_топика> <настройки_сериализации> } } } ``` Подробнее в [документации провайдера Terraform](../../../../terraform/resources/datatransfer_endpoint.md). - API {#api} * `securityGroups` — группы безопасности для сетевого трафика, правила которых применятся к ВМ и кластерам без изменения их настроек. Подробнее читайте в разделе [Сеть в Yandex Data Transfer](../../../concepts/network.md). * `connection.clusterId` — идентификатор кластера, к которому необходимо подключиться. * `auth` — метод аутентификации. Используется при подключении к хостам-брокерам: * `sasl` — аутентификация с помощью SASL. Необходимо указать следующие параметры: * `user` — имя учетной записи, от имени которой сервис Data Transfer будет подключаться к топику. * `password.raw` — пароль учетной записи (в текстовом виде). * `mechanism` — механизм хеширования. * `noAuth` — без аутентификации. {% endlist %} ### Пользовательская инсталляция {#on-premise} Подключение к кластеру Apache Kafka® с явным указанием сетевых адресов и портов хостов-брокеров. {% list tabs group=instructions %} - Консоль управления {#console} * **Тип подключения** — выберите вариант подключения к базе данных: * **Ручная настройка** — позволяет задать настройки подключения вручную. Выберите тип инсталляции **Пользовательская инсталляция** и задайте настройки: * **URL-адреса брокеров** — укажите IP-адреса или FQDN хостов-брокеров. Если номер порта Apache Kafka® отличается от стандартного, укажите его через двоеточие после имени хоста: ```text :<номер_порта> ``` * **SSL** — использовать шифрование для защиты соединения. * **PEM-сертификат** — если требуется шифрование передаваемых данных, например для соответствия требованиям PCI DSS, загрузите файл [сертификата](../../../../managed-kafka/operations/connect/index.md#get-ssl-cert) или добавьте его содержимое в текстовом виде. {% note warning %} Если не добавить сертификат, трансфер может [завершиться ошибкой](../../../troubleshooting/index.md#failed-to-connect). {% endnote %} * **Сетевой интерфейс для эндпоинта** — выберите или [создайте](../../../../vpc/operations/subnet-create.md) подсеть в нужной [зоне доступности](../../../../overview/concepts/geo-scope.md). Трансфер будет использовать эту подсеть для доступа к приемнику. Если значение в этом поле задано для обоих эндпоинтов, то обе подсети должны быть размещены в одной зоне доступности. * **Аутентификация** — выберите тип: `SASL` или `Без аутентификации`. При выборе значения `SASL`: * **Пользователь** — укажите имя учетной записи, от имени которой сервис Data Transfer будет подключаться к топику. * **Пароль** — укажите пароль учетной записи. * **Механизм** — выберите механизм хеширования: SHA 256 или SHA 512. * **Connection Manager** — позволяет подключиться к базе данных через [Yandex Connection Manager](../../../../metadata-hub/quickstart/connection-manager.md): * Выберите каталог, в котором создано подключение Connection Manager. * Выберите тип инсталляции **Пользовательская инсталляция** и задайте настройки: * **Подключение** — выберите подключение в Connection Manager или создайте его. * **Идентификатор подсети** — выберите или [создайте](../../../../vpc/operations/subnet-create.md) подсеть в нужной [зоне доступности](../../../../overview/concepts/geo-scope.md). Трансфер будет использовать эту подсеть для доступа к базе данных. Если значение в этом поле задано для обоих эндпоинтов, то обе подсети должны быть размещены в одной зоне доступности. * В блоке **Топик** выберите способ указания топика: * **Полное имя топика** — введите полное имя топика. Если вы не хотите разбивать поток событий на независимые очереди по таблицам, включите настройку **Сохранять порядок транзакций**. * **Префикс топика** — введите префикс топика. * **Группы безопасности** — выберите облачную сеть для размещения эндпоинта и группы безопасности для сетевого трафика. Это позволит применить к ВМ и кластерам в выбранной сети указанные правила групп безопасности без изменения настроек этих ВМ и кластеров. Подробнее читайте в разделе [Сеть в Yandex Data Transfer](../../../concepts/network.md). * В блоке **Настройки сериализации** выберите тип [сериализации](../../../concepts/serializer.md). Подробнее читайте в разделе [Настройки сериализации](kafka.md#serializer). * При необходимости задайте **Дополнительные настройки** эндпоинта: * Укажите конфигурацию топиков в формате **Имя конфигурации**-**Значение конфигурации**. * Выберите тип сжатия. - Terraform {#tf} * Тип эндпоинта — `kafka_target`. * `security_groups` — [группы безопасности](../../../../vpc/concepts/security-groups.md) для сетевого трафика. Правила групп безопасности применяются к трансферу. Они позволяют открыть сетевой доступ с ВМ трансфера к хостам-брокерам. Подробнее читайте в разделе [Сеть в Yandex Data Transfer](../../../concepts/network.md). Группы безопасности должны принадлежать той же сети, что и подсеть `subnet_id`, если она указана. {% note info %} В Terraform сеть для групп безопасности задавать не нужно. {% endnote %} * `connection.on_premise` — параметры подключения к хостам-брокерам: * `broker_urls` — IP-адреса или FQDN хостов-брокеров. Если номер порта Apache Kafka® отличается от стандартного, укажите его через двоеточие после имени хоста: ```text :<номер_порта> ``` * `tls_mode` — параметры шифрования передаваемых данных, если оно требуется, например для соответствия требованиям PCI DSS. * `disabled` — отключено. * `enabled` — включено. * `ca_certificate` — сертификат CA. {% note warning %} Если не добавить сертификат, трансфер может [завершиться ошибкой](../../../troubleshooting/index.md#failed-to-connect). {% endnote %} * `subnet_id` — идентификатор подсети, в которой находятся хосты-брокеры. Трансфер будет использовать эту подсеть для доступа к ним. * `auth` — метод аутентификации. Используется при подключении к хостам-брокерам: * `sasl` — аутентификация с помощью SASL. * `no_auth` — без аутентификации. Пример структуры конфигурационного файла: ```hcl resource "yandex_datatransfer_endpoint" "<имя_эндпоинта_в_Terraform>" { name = "<имя_эндпоинта>" settings { kafka_target { security_groups = ["<список_идентификаторов_групп_безопасности>"] connection { on_premise { broker_urls = ["<список_IP-адресов_или_FQDN_хостов-брокеров>"] subnet_id = "<идентификатор_подсети_c_хостами-брокерами>" } } auth = { <метод_аутентификации> } <настройки_топика> <настройки_сериализации> } } } ``` Подробнее в [документации провайдера Terraform](../../../../terraform/resources/datatransfer_endpoint.md). - API {#api} * `securityGroups` — группы безопасности для сетевого трафика, правила которых применятся к ВМ и кластерам без изменения их настроек. Подробнее читайте в разделе [Сеть в Yandex Data Transfer](../../../concepts/network.md). * `connection.onPremise` — параметры подключения к хостам-брокерам: * `brokerUrls` — IP-адреса или FQDN хостов-брокеров. Если номер порта Apache Kafka® отличается от стандартного, укажите его через двоеточие после имени хоста: ```text :<номер_порта> ``` * `tlsMode` — параметры шифрования передаваемых данных, если оно требуется, например для соответствия требованиям PCI DSS. * `disabled` — отключено. * `enabled` — включено. * `caCertificate` — сертификат CA. {% note warning %} Если не добавить сертификат, трансфер может [завершиться ошибкой](../../../troubleshooting/index.md#failed-to-connect). {% endnote %} * `subnetId` — идентификатор подсети, в которой находятся хосты-брокеры. Трансфер будет использовать эту подсеть для доступа к ним. * `auth` — метод аутентификации. Используется при подключении к хостам-брокерам: * `sasl` — аутентификация с помощью SASL. Необходимо указать следующие параметры: * `user` — имя учетной записи, от имени которой сервис Data Transfer будет подключаться к топику. * `password.raw` — пароль учетной записи (в текстовом виде). * `mechanism` — механизм хеширования. * `noAuth` — без аутентификации. {% endlist %} ### Настройки топика Apache Kafka® {#kafka-settings} {% list tabs group=instructions %} - Консоль управления {#console} * **Топик**: * **Полное имя топика** — укажите имя топика, в который будут отправляться сообщения. Выберите **Сохранять порядок транзакций**, чтобы не разбивать поток событий на независимые очереди по таблицам. * **Префикс топика** — укажите префикс топика, аналог настройки `Debezium database.server.name`. Сообщения будут отправляться в топик с именем `<префикс_топика>.<схема>.<имя_таблицы>`. - Terraform {#tf} Укажите в блоке `topic_settings` одну из опций отправки сообщений в топик: * `topic` — укажите параметры в этом блоке, чтобы отправлять все сообщения в один топик: * `topic_name` — имя топика, в который будут отправляться сообщения. * `save_tx_order` — опция, позволяющая сохранять порядок транзакций. Укажите значение `true`, чтобы не разбивать поток событий на независимые очереди по таблицам. * `topic_prefix` — укажите префикс, чтобы отправлять сообщения в разные топики с заданным префиксом. Это аналог настройки Debezium `database.server.name`. Сообщения будут отправляться в топик с именем `<префикс_топика>.<схема>.<имя_таблицы>`. - API {#api} Укажите в поле `topicSettings` одну из опций отправки сообщений в топик: * `topic` — укажите параметры в этом поле, чтобы отправлять все сообщения в один топик: * `topicName` — имя топика, в который будут отправляться сообщения. * `saveTxOrder` — опция, позволяющая сохранять порядок транзакций. Укажите значение `true`, чтобы не разбивать поток событий на независимые очереди по таблицам. * `topicPrefix` — укажите префикс, чтобы отправлять сообщения в разные топики с заданным префиксом. Это аналог настройки Debezium `database.server.name`. Сообщения будут отправляться в топик с именем `<префикс_топика>.<схема>.<имя_таблицы>`. {% endlist %} Yandex Data Transfer поддерживает CDC-режим для трансферов из баз данных PostgreSQL, MySQL® в Apache Kafka®. При этом данные в приемник попадают в формате Debezium. Подробнее о CDC-режиме читайте в разделе [Захват изменения данных](../../../concepts/cdc.md). ### Настройки сериализации {#serializer} {% list tabs group=instructions %} - Консоль управления {#console} * В блоке **Настройки сериализации** выберите тип [сериализации](../../../concepts/serializer.md): * **Auto** — автоматическая сериализация. * **Debezium** — сериализация по стандартам Debezium: * **Yandex Schema Registry**: * Выберите схему сообщения (формат данных). * Выберите из списка [пространство имен](../../../../metadata-hub/operations/list-name-space.md) реестра схем Yandex Schema Registry. * **On Premise**: * Выберите схему ключа сообщения (соответствует Debezium-параметру `key.converter`). Значение по умолчанию соответствует Debezium-параметрам `key.converter.schemas.enable=true` и `key.converter=org.apache.kafka.connect.json.JsonConverter`. При выборе значения **Отключена** Debezium-параметр `key.converter.schemas.enable` принимает значение `false`. * Выберите схему значения сообщения (соответствует Debezium-параметру `value.converter`). Значение по умолчанию соответствует Debezium-параметрам `value.converter.schemas.enable=true` и `value.converter=org.apache.kafka.connect.json.JsonConverter`. При выборе значения **Отключена** Debezium-параметр `value.converter.schemas.enable` принимает значение `false`. * При выборе значений **Avro (посредством Schema Registry)**, **JSON (посредством Schema Registry)**, **Protobuf (посредством Schema Registry)** для схем ключа и значения сообщения укажите параметры доступа к внешнему сервису Confluent Schema Registry: * **URL** — URL сервиса. * (Опционально) **Имя пользователя**. * (Опционально) **Пароль**. * (Опционально) **CA сертификат** — загрузите SSL-сертификат для подключения. Чтобы пропустить проверку сертификата, не задавайте его. * При необходимости задайте [**Настройки сериализации Debezium**](../../../concepts/serializer.md#debezium) в формате `Параметр`-`Значение`. Если вы хотите использовать JSON-схемы в Yandex Schema Registry, сохраняя [совместимость схем при добавлении и удалении опциональных полей](../../../../metadata-hub/concepts/schema-registry-content-model.md#optional-parameters-compatibility-solution), укажите следующие настройки: **Настройки сериализации** — **Debezium** — **On Premise** * Чтобы использовать Schema Registry для ключей, выберите **Схема ключа сообщения** — **JSON (посредством Schema Registry)**. Чтобы использовать Schema Registry для значений, выберите **Схема значения сообщения** — **JSON (посредством Schema Registry)**. Укажите параметры доступа к Schema Registry: * **URL** — `https://<идентификатор_пространства_имен>.schema-registry.yandexcloud.net:443`. {% note warning %} В пространстве имен должна быть [выставлена](../../../../metadata-hub/operations/update-name-space.md) **Политика проверок совместимости для JSON** — `optional friendly`. {% endnote %} * **Имя пользователя** — `api-key`. * **Пароль** — значение [API-ключа](../../../../iam/concepts/authorization/api-key.md) с ограниченной областью действия для подключения к Schema Registry. Чтобы получить значение: 1. Создайте API-ключ с ограниченной областью действия и поместите его в локальную переменную `SECRET`: ```bash yc iam api-key create --folder-id <идентификатор_каталога> \ --service-account-name <имя_сервисного_аккаунта_для_работы_со_Schema_Registry> \ --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. Выведите в терминал значение переменной `SECRET`: ```bash echo $SECRET ``` * В блоке **Настройки сериализации Debezium**: * Чтобы генерировать закрытую схему для ключей, добавьте параметр `key.converter.dt.json.generate.closed.content.schema` со значением `true`. * Чтобы генерировать закрытую схему для значений, добавьте параметр `value.converter.dt.json.generate.closed.content.schema` со значением `true`. - Terraform {#tf} Укажите в блоке `serializer` выбранный тип [сериализации](../../../concepts/serializer.md): * `serializer_auto`. * `serializer_json`. * `serializer_debezium`. Для этого типа можно указать параметры [сериализации Debezium](../../../concepts/serializer.md#debezium), задав их в блоке `serializer_debezium.serializer_parameters` в виде пар `key`/`value`. Если вы хотите использовать JSON-схемы в Yandex Schema Registry, сохраняя [совместимость схем при добавлении и удалении опциональных полей](../../../../metadata-hub/concepts/schema-registry-content-model.md#optional-parameters-compatibility-solution), добавьте в конфигурационный файл блок `serializer` с описанием настроек сериализации. Чтобы генерировать закрытую схему для ключей, добавьте в блок `serializer` переменную `key.converter.dt.json.generate.closed.content.schema` со значением `true`. Чтобы генерировать закрытую схему для значений, добавьте в блок `serializer` переменную `value.converter.dt.json.generate.closed.content.schema` со значением `true`. ```hcl resource "yandex_datatransfer_endpoint" "<имя_эндпоинта_в_Terraform>" { ... 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:<значение_API-ключа>" } } } } } } ``` Где: * `URL_пространства_имен_Schema_Registry` — эндпоинт пространства имен Schema Registry. Вы можете скопировать эндпоинт из подсказки по подключению к пространству имен Schema Registry на вкладке **Debezium**, в параметре `value.converter.schema.registry.url`. {% note warning %} В пространстве имен должна быть [выставлена](../../../../metadata-hub/operations/update-name-space.md) **Политика проверок совместимости для JSON** — `optional friendly`. {% endnote %} * `значение_API-ключа` — значение [API-ключа](../../../../iam/concepts/authorization/api-key.md) с ограниченной областью действия для подключения к Schema Registry. Чтобы получить значение: 1. Создайте API-ключ с ограниченной областью действия и поместите его в локальную переменную `SECRET`: ```bash yc iam api-key create --folder-id <идентификатор_каталога> \ --service-account-name <имя_сервисного_аккаунта_для_работы_со_Schema_Registry> \ --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. Выведите в терминал значение переменной `SECRET`: ```bash echo $SECRET ``` 1. Скопируйте полученное значение и вставьте его в параметр `value` в конфигурационном файле. - API {#api} Укажите в блоке `serializer` выбранный тип [сериализации](../../../concepts/serializer.md): * `serializerAuto`. * `serializerJson`. * `serializerDebezium`. Для этого типа можно указать параметры [сериализации Debezium](../../../concepts/serializer.md#debezium), задав их в поле `serializerDebezium.serializerParameters` в виде пар `key`/`value`. Если вы хотите использовать JSON-схемы в Yandex Schema Registry, сохраняя [совместимость схем при добавлении и удалении опциональных полей](../../../../metadata-hub/concepts/schema-registry-content-model.md#optional-parameters-compatibility-solution), добавьте в тело запроса объект `serializer` с описанием настроек сериализации. Чтобы генерировать закрытую схему для ключей, добавьте в объект `serializer` переменную `key.converter.dt.json.generate.closed.content.schema` со значением `true`. Чтобы генерировать закрытую схему для значений, добавьте в объект `serializer` переменную `value.converter.dt.json.generate.closed.content.schema` со значением `true`. ```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:<значение_API-ключа>" } ] } } ``` Где: * `URL_пространства_имен_Schema_Registry` — эндпоинт пространства имен Schema Registry. Вы можете скопировать эндпоинт из подсказки по подключению к пространству имен Schema Registry на вкладке **Debezium**, в параметре `value.converter.schema.registry.url`. {% note warning %} В пространстве имен должна быть [выставлена](../../../../metadata-hub/operations/update-name-space.md) **Политика проверок совместимости для JSON** — `optional friendly`. {% endnote %} * `значение_API-ключа` — значение [API-ключа](../../../../iam/concepts/authorization/api-key.md) с ограниченной областью действия для подключения к Schema Registry. Чтобы получить значение: 1. Создайте API-ключ с ограниченной областью действия и поместите его в локальную переменную `SECRET`: ```bash yc iam api-key create --folder-id <идентификатор_каталога> \ --service-account-name <имя_сервисного_аккаунта_для_работы_со_Schema_Registry> \ --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. Выведите в терминал значение переменной `SECRET`: ```bash echo $SECRET ``` 1. Скопируйте полученное значение и вставьте его в параметр `value` в конфигурационном файле. {% endlist %} ### Дополнительные настройки {#additional-settings} {% list tabs group=instructions %} - Консоль управления {#console} Вы можете указать [параметры конфигурации топика](https://docs.confluent.io/platform/current/installation/configuration/topic-configs.html), которые будут применяться при создании новых топиков. Укажите параметр и одно из его допустимых значений: например, `cleanup.policy` и `compact`. {% endlist %} После настройки источника и приемника данных [создайте и запустите трансфер](../../transfer.md#create).