# Поставка данных из очереди Data Streams в Managed Service for Apache Kafka® # Поставка данных из очереди Data Streams в Managed Service for Apache Kafka® с помощью Data Transfer С помощью сервиса Data Transfer можно поставлять данные из потока Data Streams в кластер Managed Service for Apache Kafka®. Чтобы перенести данные: 1. [Подготовьте поток данных Data Streams](#prepare-source). 1. [Подготовьте и активируйте трансфер](#prepare-transfer). 1. [Проверьте работоспособность трансфера](#verify-transfer). Если созданные ресурсы вам больше не нужны, [удалите их](#clear-out). ## Необходимые платные ресурсы {#paid-resources} * База данных Managed Service for YDB ([тарифы Managed Service for YDB](../../ydb/pricing/index.md)). Стоимость зависит от режима использования: * Для бессерверного режима — оплачиваются операции с данными, объем хранимых данных и резервных копий. * Для режима с выделенными инстансами — оплачивается использование выделенных БД вычислительных ресурсов, объем хранилища и резервные копии. * Сервис Data Streams ([тарифы Data Streams](../../data-streams/pricing.md)). Стоимость зависит от режима тарификации: * [По выделенным ресурсам](../../data-streams/pricing.md#rules) — оплачивается фиксированная почасовая ставка за установленный лимит пропускной способности и срок хранения сообщений, а также дополнительно количество единиц фактически записанных данных. * [По фактическому использованию](../../data-streams/pricing.md#on-demand) (On-demand) — оплачиваются выполненные операции записи и чтения данных, объем считанных/записанных данных, а также объем фактически используемого хранилища для сообщений, по которым не истек срок хранения. * Кластер Managed Service for Apache Kafka®: выделенные хостам вычислительные ресурсы, объем хранилища и резервных копий ([тарифы Managed Service for Apache Kafka®](../../managed-kafka/pricing.md)). * Публичные IP-адреса, если для хостов кластера включен публичный доступ ([тарифы Virtual Private Cloud](../../vpc/pricing.md)). * Каждый трансфер: использование вычислительных ресурсов и количество переданных строк данных ([тарифы Data Transfer](../pricing.md)). ## Перед началом работы {#before-you-begin} Подготовьте инфраструктуру поставки данных: {% list tabs group=instructions %} - Вручную {#manual} 1. [Создайте базу данных Managed Service for YDB](../../ydb/operations/manage-databases.md) любой подходящей конфигурации. 1. [Создайте кластер Managed Service for Apache Kafka®](../../managed-kafka/operations/cluster-create.md) любой подходящей конфигурации c хостами в публичном доступе. {% note info %} Публичный доступ к хостам кластера нужен, если вы планируете подключаться к кластеру через интернет. Этот вариант подключения более простой, и его рекомендуется использовать для прохождения руководства. К хостам без публичного доступа тоже можно подключиться, но только с виртуальных машин Yandex Cloud, расположенных в той же облачной сети, что и кластер. {% endnote %} 1. [Создайте в кластере Managed Service for Apache Kafka® топик](../../managed-kafka/operations/cluster-topics.md#create-topic) с именем `sensors`. 1. [Создайте в кластере Managed Service for Apache Kafka® пользователя](../../managed-kafka/operations/cluster-accounts.md#create-account) с именем `mkf-user` и правами доступа `ACCESS_ROLE_PRODUCER` и `ACCESS_ROLE_CONSUMER` к созданному топику. - Terraform {#tf} 1. Если у вас еще нет Terraform, [установите его](../../tutorials/infrastructure-management/terraform-quickstart.md#install-terraform). 1. [Получите данные для аутентификации](../../tutorials/infrastructure-management/terraform-quickstart.md#get-credentials). Вы можете добавить их в переменные окружения или указать далее в файле с настройками провайдера. 1. [Настройте и инициализируйте провайдер](../../tutorials/infrastructure-management/terraform-quickstart.md#configure-provider). Чтобы не создавать конфигурационный файл с настройками провайдера вручную, [скачайте его](https://github.com/yandex-cloud-examples/yc-terraform-provider-settings/blob/main/provider.tf). 1. Поместите конфигурационный файл в отдельную рабочую директорию и [укажите значения параметров](../../tutorials/infrastructure-management/terraform-quickstart.md#configure-provider). Если данные для аутентификации не были добавлены в переменные окружения, укажите их в конфигурационном файле. 1. Скачайте в ту же рабочую директорию файл конфигурации [yds-to-kafka.tf](https://github.com/yandex-cloud-examples/yc-data-transfer-from-yds-to-kafka/blob/main/yds-to-kafka.tf). В этом файле описаны: * [сеть](../../vpc/concepts/network.md#network); * [подсеть](../../vpc/concepts/network.md#subnet); * [группа безопасности](../../vpc/concepts/security-groups.md) и правила, необходимые для подключения к кластеру Managed Service for Apache Kafka®; * база данных Managed Service for YDB; * кластер Managed Service for Apache Kafka®; * топик Managed Service for Apache Kafka® с именем `sensors`; * пользователь Managed Service for Apache Kafka® с правами доступа `ACCESS_ROLE_PRODUCER`, `ACCESS_ROLE_CONSUMER` к топику `sensors`; * трансфер. 1. Укажите в файле `yds-to-kafka.tf` значения параметров: * `mkf_version` — версия кластера Apache Kafka®; * `ydb_name` — имя базы данных Managed Service for YDB; * `mkf_user_name` — имя пользователя в кластере Managed Service for Apache Kafka®; * `mkf_user_password` — пароль пользователя в кластере Managed Service for Apache Kafka®; * `transfer_enabled` — значение `0`, чтобы не создавать трансфер до [создания эндпоинтов вручную](#prepare-transfer). 1. Проверьте корректность файлов конфигурации Terraform с помощью команды: ```bash terraform validate ``` Если в файлах конфигурации есть ошибки, Terraform на них укажет. 1. Создайте необходимую инфраструктуру: 1. Выполните команду для просмотра планируемых изменений: ```bash terraform plan ``` Если конфигурации ресурсов описаны верно, в терминале отобразится список изменяемых ресурсов и их параметров. Это проверочный этап: ресурсы не будут изменены. 1. Если вас устраивают планируемые изменения, внесите их: 1. Выполните команду: ```bash terraform apply ``` 1. Подтвердите изменение ресурсов. 1. Дождитесь завершения операции. В указанном каталоге будут созданы все требуемые ресурсы. Проверить появление ресурсов и их настройки можно в [консоли управления](https://kz.console.yandex.cloud). {% endlist %} ## Создайте поток данных Data Streams {#prepare-source} [Создайте поток данных Data Streams](../../data-streams/quickstart/create-stream.md). ## Подготовьте и активируйте трансфер {#prepare-transfer} 1. [Создайте эндпоинт для источника Data Streams](../operations/endpoint/index.md#create). * **Тип базы данных** — `Yandex Data Streams`. * **Параметры эндпоинта**: * **Настройки подключения**: * **База данных** — выберите базу данных Managed Service for YDB из списка. * **Поток** — укажите имя потока Data Streams. * **Сервисный аккаунт** — выберите или создайте сервисный аккаунт с ролью `yds.editor`. * **Расширенные настройки**: * **Правила конвертации** — `JSON`. * **Схема данных** — `JSON-спецификация`: Заполните схему данных: {% cut "Схема данных" %} ```json [ { "name": "device_id", "type": "string" }, { "name": "datetime", "type": "datetime" }, { "name": "latitude", "type": "double" }, { "name": "longitude", "type": "double" }, { "name": "altitude", "type": "double" }, { "name": "speed", "type": "double" }, { "name": "battery_voltage", "type": "any" }, { "name": "cabin_temperature", "type": "double" }, { "name": "fuel_level", "type": "any" } ] ``` {% endcut %} 1. [Создайте эндпоинт для приемника Managed Service for Apache Kafka®](../operations/endpoint/index.md#create). * **Тип базы данных** — `Kafka`. * **Параметры эндпоинта**: * **Настройки подключения**: * **Тип подключения** — выберите `Кластер Managed Service for Apache Kafka`. * **Кластер Managed Service for Apache Kafka** — выберите кластер Managed Service for Apache Kafka® из списка. * **Аутентификация** — выберите **SASL**. * **Имя пользователя** — введите имя пользователя в кластере Managed Service for Apache Kafka®. * **Пароль** — введите пароль пользователя в кластере Managed Service for Apache Kafka®. * **Топик** — выберите **Полное имя топика**. * **Полное имя топика** — введите имя топика в кластере Managed Service for Apache Kafka®. 1. Создайте трансфер: {% list tabs group=instructions %} - Вручную {#manual} 1. [Создайте трансфер](../operations/transfer.md#create) типа **Репликация**, использующий созданные эндпоинты. 1. [Активируйте](../operations/transfer.md#activate) его. - Terraform {#tf} 1. Укажите в файле `yds-to-kafka.tf` значения переменных: * `source_endpoint_id` — идентификатор эндпоинта для источника; * `target_endpoint_id` — идентификатор эндпоинта для приемника; * `transfer_enabled` – значение `1` для создания трансфера. 1. Проверьте корректность файлов конфигурации Terraform с помощью команды: ```bash terraform validate ``` Если в файлах конфигурации есть ошибки, Terraform на них укажет. 1. Создайте необходимую инфраструктуру: 1. Выполните команду для просмотра планируемых изменений: ```bash terraform plan ``` Если конфигурации ресурсов описаны верно, в терминале отобразится список изменяемых ресурсов и их параметров. Это проверочный этап: ресурсы не будут изменены. 1. Если вас устраивают планируемые изменения, внесите их: 1. Выполните команду: ```bash terraform apply ``` 1. Подтвердите изменение ресурсов. 1. Дождитесь завершения операции. Трансфер активируется автоматически. {% endlist %} ## Проверьте работоспособность трансфера {#verify-transfer} 1. Дождитесь перехода трансфера в статус **Реплицируется**. 1. [Отправьте в поток Data Streams](../../data-streams/operations/aws-cli/send.md) тестовые данные: ```json { "device_id":"iv9a94th6rzt********", "datetime":"2020-06-05T17:27:00", "latitude":"55.70329032", "longitude":"37.65472196", "altitude":"427.5", "speed":"0", "battery_voltage":"23.5", "cabin_temperature":"17", "fuel_level":null } ``` 1. Убедитесь, что данные перенеслись в топик `sensors` кластера Managed Service for Apache Kafka®: 1. [Получите SSL-сертификат](../../managed-kafka/operations/connect/index.md#get-ssl-cert) для подключения к кластеру Managed Service for Apache Kafka®. 1. [Установите](../../managed-kafka/operations/connect/clients.md#bash-zsh) утилиту `kafkacat`. 1. [Запустите](../../managed-kafka/operations/connect/clients.md#with-ssl_1) команду получения сообщений из топика. ## Удалите созданные ресурсы {#clear-out} {% note info %} Перед тем как удалить созданные ресурсы, [деактивируйте трансфер](../operations/transfer.md#deactivate). {% endnote %} Чтобы снизить потребление ресурсов, которые вам не нужны, удалите их: 1. [Удалите трансфер](../operations/transfer.md#delete). 1. [Удалите эндпоинты](../operations/endpoint/index.md#delete) для источника и приемника. 1. Если при создании эндпоинта для источника вы создавали сервисный аккаунт, [удалите его](../../iam/operations/sa/delete.md). 1. Остальные ресурсы удалите в зависимости от способа их создания: {% list tabs group=instructions %} - Вручную {#manual} 1. [Удалите кластер Managed Service for Apache Kafka®](../../managed-kafka/operations/cluster-delete.md). 1. [Удалите базу данных Managed Service for YDB](../../ydb/operations/manage-databases.md#delete-db).. - Terraform {#tf} 1. В терминале перейдите в директорию с планом инфраструктуры. {% note warning %} Убедитесь, что в директории нет Terraform-манифестов с ресурсами, которые вы хотите сохранить. Terraform удаляет все ресурсы, которые были созданы с помощью манифестов в текущей директории. {% endnote %} 1. Удалите ресурсы: 1. Выполните команду: ```bash terraform destroy ``` 1. Подтвердите удаление ресурсов и дождитесь завершения операции. Все ресурсы, которые были описаны в Terraform-манифестах, будут удалены. {% endlist %}