[Документация 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. [Подготовьте базу данных Apache Kafka®](#prepare) к трансферу. 1. [Настройте эндпоинт-источник](#endpoint-settings) в Yandex Data Transfer. 1. [Настройте один из поддерживаемых приемников данных](#supported-targets). 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. Поставка данных — процесс доставки произвольных данных в целевые хранилища. Процесс поставки включает извлечение данных из очереди и их десериализацию с последующей трансформацией данных в формат целевого хранилища. * [Apache Kafka® в ClickHouse®](../../../tutorials/mkf-to-mch.md); * [Apache Kafka® в PostgreSQL](../../../tutorials/mkf-to-mpg.md); * [Apache Kafka® в Greenplum®](../../../tutorials/managed-kafka-to-greenplum.md); * [Apache Kafka® в MongoDB](../../../tutorials/mkf-to-mmg.md); * [Apache Kafka® в MySQL®](../../../tutorials/mkf-to-mmy.md); * [Apache Kafka® в OpenSearch](../../../tutorials/mkf-to-mos.md); Подробное описание возможных сценариев передачи данных в Yandex Data Transfer читайте в разделе [Практические руководства](../../../tutorials/index.md). ## Подготовка базы данных источника {#prepare} {% list tabs %} - Managed Service for Apache Kafka® [Создайте пользователя](../../../../managed-kafka/operations/cluster-accounts.md#create-account) с ролью `ACCESS_ROLE_CONSUMER` на топике-источнике. - Apache Kafka® 1. Если вы не планируете использовать для подключения к внешнему кластеру [сервис Cloud Interconnect](../../../../interconnect/concepts/index.md) или [VPN](../../../../glossary/vpn.md), разрешите подключения к такому кластеру из интернета с [IP-адресов, используемых сервисом Data Transfer](../../../../overview/concepts/public-ips.md#virtual-private-cloud). Подробнее о настройке сети для работы с внешними ресурсами читайте в [концепции](../../../concepts/network.md#source-external). 1. [Настройте доступ к кластеру-источнику из Yandex Cloud](../../../concepts/network.md#source-external). 1. [Настройте права доступа](https://kafka.apache.org/42/security/authorization-and-acls/#examples) пользователя к нужному топику. 1. Выдайте права `READ` консьюмер-группе, идентификатор которой совпадает с идентификатором трансфера. ```text bin/kafka-acls --bootstrap-server localhost:9092 \ --command-config adminclient-configs.conf \ --add \ --allow-principal User:username \ --operation Read \ --group <идентификатор_трансфера> ``` 1. (Опционально) Чтобы использовать авторизацию по логину и паролю, [настройте SASL-аутентификацию](https://kafka.apache.org/42/security/authentication-using-sasl/). {% endlist %} ## Настройка эндпоинта-источника Apache Kafka® При [создании](../index.md#create) или [изменении](../index.md#update) эндпоинта вы можете задать: * Настройки подключения к [кластеру Yandex Managed Service for Apache Kafka®](#managed-service) или [пользовательской инсталляции](#on-premise), в т. ч. на базе виртуальных машин Yandex Compute Cloud. Эти параметры обязательные. * [Расширенные настройки](#additional-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). - Terraform {#tf} * Тип эндпоинта — `kafka_source`. * `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` — без аутентификации. * `topic_names` — список имен топиков, к которым необходимо подключиться. Пример структуры конфигурационного файла: ```hcl resource "yandex_datatransfer_endpoint" "<имя_эндпоинта_в_Terraform>" { name = "<имя_эндпоинта>" settings { kafka_source { security_groups = ["<список_идентификаторов_групп_безопасности>"] connection { cluster_id = "<идентификатор_кластера>" } auth { <метод_аутентификации> } topic_names = ["<список_имен_топиков>"] <расширенные_настройки_эндпоинта> } } } ``` Подробнее в [документации провайдера 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` — без аутентификации. * `topicNames` — список имен топиков, к которым необходимо подключиться. {% 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). - Terraform {#tf} * Тип эндпоинта — `kafka_source`. * `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` — без аутентификации. * `topic_names` — список имен топиков, к которым необходимо подключиться. Пример структуры конфигурационного файла: ```hcl resource "yandex_datatransfer_endpoint" "<имя_эндпоинта_в_Terraform>" { name = "<имя_эндпоинта>" settings { kafka_source { security_groups = ["<список_идентификаторов_групп_безопасности>"] connection { on_premise { broker_urls = ["<список_IP-адресов_или_FQDN_хостов-брокеров>"] subnet_id = "<идентификатор_подсети_c_хостами-брокерами>" } } auth = { <метод_аутентификации> } topic_names = ["<список_имен_топиков>"] <расширенные_настройки_эндпоинта> } } } ``` Подробнее в [документации провайдера 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` — без аутентификации. * `topicNames` — список имен топиков, к которым необходимо подключиться. {% endlist %} ### Расширенные настройки {#additional-settings} В расширенных настройках вы можете задать правила трансформации и правила конвертации. Данные обрабатываются в следующем порядке: 1. **Трансформация**. Данные в формате JSON передаются функции [Yandex Cloud Functions](../../../../functions/index.md). В теле функции содержатся метаинформация и необработанные данные, которые переданы в очередь. С помощью функции данные обрабатываются и возвращаются в Data Transfer. 1. **Конвертация**. Выполняется парсинг, с помощью которого данные подготавливаются для передачи приемнику. Если не заданы правила трансформации, то парсинг применяется к необработанным данным из очереди. Если не заданы правила конвертации, то данные сразу передаются в приемник. {% list tabs group=instructions %} - Консоль управления {#console} * **Правила трансформации**: * **Функция обработки** — выберите одну из функций, созданных в сервисе Cloud Functions. * **Сервисный аккаунт** — выберите или [создайте](../../../../iam/operations/sa/create.md) сервисный аккаунт, от имени которого будет запускаться функция обработки. * **Количество попыток** — укажите количество попыток вызова функции обработки. * **Размер буфера для отправки** — укажите размер буфера (в байтах), при заполнении которого данные будут переданы функции обработки. Максимальный размер буфера — 3,5 МБ. Подробнее об ограничениях, действующих при работе с функциями в сервисе Cloud Functions, читайте в [соответствующем разделе](../../../../functions/concepts/limits.md). * **Интервал отправки** — укажите длительность интервала (в секундах), по истечении которого данные из потока должны быть переданы функции обработки. {% note info %} Если буфер заполнится или истечет интервал отправки, данные будут переданы функции обработки. {% endnote %} * **Таймаут вызова** — укажите допустимое время ожидания ответа от функции обработки (в секундах). {% note warning %} Значения в полях **Интервал отправки** и **Таймаут вызова** указываются с постфиксом `s`, например, `10s`. {% endnote %} * **Правила конвертации**: * **Формат данных** — выберите один из доступных форматов: * `JSON` — формат JSON. Для оптимизации пропускной способности поддерживается обработка сообщений JSON, записанных одной строкой — без символов перевода строки `\n`. Пример: `{"attr": "value"}` * `Парсер AuditTrails.v1` — формат логов сервиса [Audit Trails](../../../../audit-trails/index.md). * `Парсер CloudLogging` — формат логов сервиса [Cloud Logging](../../../../logging/index.md). * `Парсер Debezium CDC` — Debezium CDC. Позволяет указать настройки подключения к Schema Registry: `On Premise Schema Registry` для [Confluent Schema Registry](https://docs.confluent.io/platform/current/schema-registry/index.html), идентификатор [пространства имен](../../../../metadata-hub/operations/list-name-space.md) для `Yandex Schema Registry`. * `Парсер Raw-to-table` — позволяет сохранять сообщения целиком «как есть» в отдельную таблицу приемника. Для формата **JSON** укажите: * **Схема данных** — задайте схему в виде списка полей или загрузите файл с описанием схемы в формате JSON. {% cut "Пример задания схемы данных" %} ```json [ { "name": "request", "type": "string" } ] ``` {% endcut %} * **Использовать значение NULL в ключевых столбцах** — выберите эту опцию, чтобы разрешить значение `null` в ключевых колонках. * **Добавить неразмеченные столбцы** — выберите эту опцию, чтобы поля, отсутствующие в схеме, попадали в колонку `_rest`. * **Разэкранировать значения строк** — выберите эту опцию, чтобы убрать кавычки из строковых переменных (если этого не сделать, значения строковых полей останутся без изменений). Для **Debezium CDC** укажите: * Для `On Premise Schema Registry` — URL для Schema Registry, способ аутентификации (с указанием логина и пароля пользователя в случае использования аутентификации) и CA-сертификат. * Для `Yandex Schema Registry` — идентификатор [пространства имен](../../../../metadata-hub/operations/list-name-space.md) реестра схем. Для **Raw-to-table** укажите: * **Имя целевой таблицы** — имя таблицы, в которую сохраняются сообщения. По умолчанию совпадает с именем топика. * **Сохранять ключ сообщения** — выберите эту опцию, чтобы записывать ключ сообщения в отдельный столбец. * **Тип ключа** — выберите подходящий формат для записи ключа в таблицу. * **Тип значения** — выберите подходящий формат для записи значения в таблицу. * **Дополнительные параметры**: * **Сохранять время записи** — выберите опцию, чтобы сохранять время записи сообщения в отдельном столбце. * **Сохранять заголовки сообщения** — выберите опцию, чтобы сохранять заголовки сообщения в отдельном столбце. * **Суффикс имени таблицы DLQ** — укажите суффикс, который будет использоваться для именования отдельной таблицы с недоставленными сообщениями по схеме `<имя_таблицы_сообщений>_<суффикс>`. По умолчанию применяется суффикс `_dlq`. - Terraform {#tf} * `transformer` — правила трансформации. * `parser` — правила конвертации, которые определяются одним из выбранных парсеров: * `audit_trails_v1_parser` — парсер для логов сервиса [Audit Trails](../../../../audit-trails/index.md). * `cloud_logging_parser` — парсер для логов сервиса [Cloud Logging](../../../../logging/index.md). * `json_parser` или `tskv_parser` — парсеры JSON и TSKV, соответственно. Атрибуты, задающие параметры их работы, одинаковы для обоих парсеров: * `data_schema` — схема данных, представленная либо в виде JSON-объекта с описанием схемы, либо в виде списка полей: * `fields` — схема в виде JSON-объекта. Объект содержит в себе массив JSON-объектов, которые описывают отдельные колонки. * `json_fields` — схема в виде списка JSON-полей. * `null_keys_allowed` — задайте значение `true` для этого атрибута, чтобы разрешить значение `null` в ключевых колонках. * `add_rest_column` — задайте значение `true` для этого атрибута, чтобы поля, отсутствующие в схеме, попадали в колонку `_rest`. * `unescape_string_values` — задайте значение `true` для этого атрибута, чтобы убрать кавычки из строковых переменных (если этого не сделать, значения строковых полей останутся без изменений). - API {#api} * `transformer` — правила трансформации: * `cloudFunction` — идентификатор функции, созданной в сервисе Cloud Functions. * `serviceAccountId` — идентификатор сервисного аккаунта, от имени которого будет запускаться функция обработки. * `numberOfRetries` — количество попыток вызова функции обработки. * `bufferSize` — размер буфера (в байтах), при заполнении которого данные будут переданы функции обработки. Максимальный размер буфера — 3,5 МБ. Подробнее об ограничениях, действующих при работе с функциями в сервисе Cloud Functions, читайте в [соответствующем разделе](../../../../functions/concepts/limits.md). * `bufferFlushInterval` — длительность интервала (в секундах), по истечении которого данные из потока должны быть переданы функции обработки. {% note info %} Если буфер заполнится или истечет интервал отправки, данные будут переданы функции обработки. {% endnote %} * `invocationTimeout` — допустимое время ожидания ответа от функции обработки (в секундах). {% note warning %} Значения для `bufferFlushInterval` и `invocationTimeout` указываются с постфиксом `s`, например, `10s`. {% endnote %} * `parser` — правила конвертации, которые определяются одним из выбранных парсеров: * `auditTrailsV1Parser` — парсер для логов сервиса [Audit Trails](../../../../audit-trails/index.md). * `cloudLoggingParser` — парсер для логов сервиса [Cloud Logging](../../../../logging/index.md). * `jsonParser` или `tskvParser` — парсеры JSON и TSKV, соответственно. Поля, задающие параметры их работы, одинаковы для обоих парсеров: * `dataSchema` — схема данных, представленная либо в виде JSON-объекта с описанием схемы, либо в виде списка полей: * `fields` — схема в виде JSON-объекта. Объект содержит в себе массив JSON-объектов, которые описывают отдельные колонки. * `jsonFields` — схема в виде списка JSON-полей. * `nullKeysAllowed` — задайте значение `true` для этого поля, чтобы разрешить значение `null` в ключевых колонках. * `addRestColumn` — задайте значение `true` для этого поля, чтобы поля, отсутствующие в схеме, попадали в колонку `_rest`. * `unescapeStringValues` — задайте значение `true` для этого поля, чтобы убрать кавычки из строковых переменных (если этого не сделать, значения строковых полей останутся без изменений). {% endlist %} ## Настройка приемника данных {#supported-targets} Настройте один из поддерживаемых приемников данных: * [PostgreSQL](../target/postgresql.md) * [MySQL®](../target/mysql.md) * [MongoDB](../target/mongodb.md) * [ClickHouse®](../target/clickhouse.md) * [Greenplum®](../target/greenplum.md) * [Yandex Object Storage](../target/object-storage.md) * [Apache Kafka®](../target/kafka.md) * [OpenSearch](../target/opensearch.md) Полный список поддерживаемых источников и приемников в Yandex Data Transfer читайте в разделе [Доступные трансферы](../../../transfer-matrix.md). После настройки источника и приемника данных [создайте и запустите трансфер](../../transfer.md#create).