Управление коннекторами

Если у вас еще нет интерфейса командной строки Yandex Cloud (CLI), установите и инициализируйте его.

По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

Чтобы запросить список коннекторов кластера, выполните команду:

yc managed-kafka connector list --cluster-name=<имя_кластера>

Результат:

+--------------+-----------+
|     NAME     | TASKS MAX |
+--------------+-----------+
| connector559 |         1 |
| ...          |           |
+--------------+-----------+

Имя кластера можно получить со списком кластеров в каталоге.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Воспользуйтесь методом Connector.list и выполните запрос, например, с помощью cURL:

   curl \
     --request GET \
     --header "Authorization: Bearer $IAM_TOKEN" \
     --url 'https://mdb.api.cloud.yandex.net/managed-kafka/v1/clusters/<идентификатор_кластера>/connectors'
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге.

3. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Клонируйте репозиторий cloudapi:

   cd ~/ && git clone --depth=1 https://github.com/yandex-cloud/cloudapi
   

   Далее предполагается, что содержимое репозитория находится в директории ~/cloudapi/.

3. Воспользуйтесь вызовом ConnectorService/List и выполните запрос, например, с помощью gRPCurl:

   grpcurl \
     -format json \
     -import-path ~/cloudapi/ \
     -import-path ~/cloudapi/third_party/googleapis/ \
     -proto ~/cloudapi/yandex/cloud/mdb/kafka/v1/connector_service.proto \
     -rpc-header "Authorization: Bearer $IAM_TOKEN" \
     -d '{
           "cluster_id": "<идентификатор_кластера>"
         }' \
     mdb.api.cloud.yandex.net:443 \
     yandex.cloud.mdb.kafka.v1.ConnectorService.List
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге.

4. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

Если у вас еще нет интерфейса командной строки Yandex Cloud (CLI), установите и инициализируйте его.

По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

Чтобы получить детальную информацию о коннекторе, выполните команду:

yc managed-kafka connector get <имя_коннектора>\
   --cluster-name=<имя_кластера>

Результат:

name: connector785
tasks_max: "1"
cluster_id: c9qbkmoiimsl********
...

Имя коннектора можно запросить со списком коннекторов в кластере, имя кластера — со списком кластеров в каталоге.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Воспользуйтесь методом Connector.get и выполните запрос, например, с помощью cURL:

   curl \
     --request GET \
     --header "Authorization: Bearer $IAM_TOKEN" \
     --url 'https://mdb.api.cloud.yandex.net/managed-kafka/v1/clusters/<идентификатор_кластера>/connectors/<имя_коннектора>'
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге, а имя коннектора — со списком коннекторов в кластере.

3. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Клонируйте репозиторий cloudapi:

   cd ~/ && git clone --depth=1 https://github.com/yandex-cloud/cloudapi
   

   Далее предполагается, что содержимое репозитория находится в директории ~/cloudapi/.

3. Воспользуйтесь вызовом ConnectorService/Get и выполните запрос, например, с помощью gRPCurl:

   grpcurl \
     -format json \
     -import-path ~/cloudapi/ \
     -import-path ~/cloudapi/third_party/googleapis/ \
     -proto ~/cloudapi/yandex/cloud/mdb/kafka/v1/connector_service.proto \
     -rpc-header "Authorization: Bearer $IAM_TOKEN" \
     -d '{
           "cluster_id": "<идентификатор_кластера>"
         }' \
     mdb.api.cloud.yandex.net:443 \
     yandex.cloud.mdb.kafka.v1.ConnectorService.Get
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге.

4. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

Если у вас еще нет интерфейса командной строки Yandex Cloud (CLI), установите и инициализируйте его.

По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

Чтобы создать коннектор MirrorMaker:

1. Посмотрите описание команды CLI для создания коннектора:

   yc managed-kafka connector-mirrormaker create --help
   

2. Создайте коннектор:

   yc managed-kafka connector-mirrormaker create <имя_коннектора> \
      --cluster-name=<имя_кластера> \
      --direction=<направление_коннектора> \
      --tasks-max=<лимит_задач> \
      --properties=<дополнительные_свойства> \
      --replication-factor=<фактор_репликации> \
      --topics=<шаблон_для_топиков> \
      --this-cluster-alias=<префикс_для_обозначения_этого_кластера> \
      --external-cluster alias=<префикс_для_обозначения_внешнего_кластера>,`
                        `bootstrap-servers=<список_FQDN_хостов-брокеров>,`
                        `security-protocol=<протокол_безопасности>,`
                        `sasl-mechanism=<механизм_шифрования>,`
                        `sasl-username=<имя_пользователя>,`
                        `sasl-password=<пароль_пользователя>,`
                        `ssl-truststore-certificates=<сертификаты_в_формате_PEM>
   

   Как получить FQDN хоста-брокера, см. в инструкции.

   Имя кластера можно получить со списком кластеров в каталоге.

   Параметр --direction принимает значение:

   • egress — если текущий кластер является кластером-источником.
   • ingress — если текущий кластер является кластером-приемником.

Чтобы создать коннектор S3 Sink:

1. Посмотрите описание команды CLI для создания коннектора:

   yc managed-kafka connector-s3-sink create --help
   

2. Создайте коннектор:

   yc managed-kafka connector-s3-sink create <имя_коннектора> \
      --cluster-name=<имя_кластера> \
      --tasks-max=<лимит_задач> \
      --properties=<дополнительные_свойства> \
      --topics=<шаблон_для_топиков> \
      --file-compression-type=<кодек_сжатия> \
      --file-max-records=<максимальное_количество_сообщений_в_файле> \
      --bucket-name=<имя_бакета> \
      --access-key-id=<идентификатор_AWS-совместимого_статического_ключа> \
      --secret-access-key=<содержимое_AWS-совместимого_статического_ключа> \
      --storage-endpoint=<эндпоинт_S3-совместимого_хранилища> \
      --region=<регион_S3-совместимого_хранилища>
   

   Имя кластера можно получить со списком кластеров в каталоге.

1. Ознакомьтесь со списком настроек коннекторов MirrorMaker и S3 Sink.

2. Откройте актуальный конфигурационный файл Terraform с планом инфраструктуры.

   О том, как создать такой файл, см. в разделе Создание кластера.

3. Чтобы создать коннектор MirrorMaker, добавьте ресурс yandex_mdb_kafka_connector с блоком настроек connector_config_mirrormaker:

   resource "yandex_mdb_kafka_connector" "<имя_коннектора>" {
     cluster_id = "<идентификатор_кластера>"
     name       = "<имя_коннектора>"
     tasks_max  = <лимит_задач>
     properties = {
       <дополнительные_свойства>
     }
     connector_config_mirrormaker {
       topics             = "<шаблон_для_топиков>"
       replication_factor = <фактор_репликации>
       source_cluster {
         alias = "<префикс_для_обозначения_кластера>"
         external_cluster {
           bootstrap_servers           = "<список_FQDN_хостов-брокеров>"
           sasl_username               = "<имя_пользователя>"
           sasl_password               = "<пароль_пользователя>"
           sasl_mechanism              = "<механизм_шифрования>"
           security_protocol           = "<протокол_безопасности>"
           ssl-truststore-certificates = "<содержимое_PEM-сертификата>"
         }
       }
       target_cluster {
         alias = "<префикс_для_обозначения_кластера>"
         this_cluster {}
       }
     }
   }
   

   Как получить FQDN хоста-брокера, см. в инструкции.

4. Чтобы создать коннектор S3 Sink, добавьте ресурс yandex_mdb_kafka_connector с блоком настроек connector_config_s3_sink:

   resource "yandex_mdb_kafka_connector" "<имя_коннектора>" {
     cluster_id = "<идентификатор_кластера>"
     name       = "<имя_коннектора>"
     tasks_max  = <лимит_задач>
     properties = {
       <дополнительные_свойства>
     }
     connector_config_s3_sink {
       topics                = "<шаблон_для_топиков>"
       file_compression_type = "<кодек_сжатия>"
       file_max_records      = <максимальное_количество_сообщений_в_файле>
       s3_connection {
         bucket_name = "<имя_бакета>"
         external_s3 {
           endpoint          = "<эндпоинт_S3-совместимого_хранилища>"
           access_key_id     = "<идентификатор_AWS-совместимого_статического_ключа>"
           secret_access_key = "<содержимое_AWS-совместимого_статического_ключа>"
         }
       }
     }
   }
   

5. Проверьте корректность настроек.

   1. В командной строке перейдите в каталог, в котором расположены актуальные конфигурационные файлы Terraform с планом инфраструктуры.

   2. Выполните команду:

      terraform validate
      

      Если в файлах конфигурации есть ошибки, Terraform на них укажет.

6. Подтвердите изменение ресурсов.

   1. Выполните команду для просмотра планируемых изменений:

      terraform plan
      

      Если конфигурации ресурсов описаны верно, в терминале отобразится список изменяемых ресурсов и их параметров. Это проверочный этап: ресурсы не будут изменены.

   2. Если вас устраивают планируемые изменения, внесите их:

      1. Выполните команду:

         terraform apply
         

      2. Подтвердите изменение ресурсов.

      3. Дождитесь завершения операции.

Подробнее см. в документации провайдера Terraform.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Чтобы создать коннектор MirrorMaker, воспользуйтесь методом Connector.create и выполните запрос, например, с помощью cURL:

   curl \
     --request POST \
     --header "Authorization: Bearer $IAM_TOKEN" \
     --url 'https://mdb.api.cloud.yandex.net/managed-kafka/v1/clusters/<идентификатор_кластера>/connectors' \
     --data '{
               "connectorSpec": {
                 "name": "<имя_коннектора>",
                 "tasksMax": "<лимит_задач>"
                 "properties": "<дополнительные_свойства_коннектора>"
                 "connectorConfigMirrormaker": {
                   <настройки_коннектора_Mirrormaker>
                 }
               }
             }'
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге.

3. Чтобы создать коннектор S3 Sink, воспользуйтесь методом Connector.create и выполните запрос, например, с помощью cURL:

   curl \
     --request POST \
     --header "Authorization: Bearer $IAM_TOKEN" \
     --url 'https://mdb.api.cloud.yandex.net/managed-kafka/v1/clusters/<идентификатор_кластера>/connectors' \
     --data '{
               "connectorSpec": {
                 "name": "<имя_коннектора>",
                 "tasksMax": "<лимит_задач>"
                 "properties": "<дополнительные_свойства_коннектора>"
                 "connectorConfigS3Sink": {
                   <настройки_коннектора_S3-Sink>
                 }
               }
             }'
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге.

4. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Клонируйте репозиторий cloudapi:

   cd ~/ && git clone --depth=1 https://github.com/yandex-cloud/cloudapi
   

   Далее предполагается, что содержимое репозитория находится в директории ~/cloudapi/.

3. Чтобы создать коннектор MirrorMaker, воспользуйтесь вызовом ConnectorService/Create и выполните запрос, например, с помощью gRPCurl:

   grpcurl \
     -format json \
     -import-path ~/cloudapi/ \
     -import-path ~/cloudapi/third_party/googleapis/ \
     -proto ~/cloudapi/yandex/cloud/mdb/kafka/v1/connector_service.proto \
     -rpc-header "Authorization: Bearer $IAM_TOKEN" \
     -d '{
           "cluster_id": "<идентификатор_кластера>",
           "connector_spec": {
             "name": "<имя_коннектора>",
             "tasks_max": {
               "value": "<лимит_задач>"
             },
             "properties": "<дополнительные_свойства_коннектора>"
             "connector_config_mirrormaker": {
               <настройки_коннектора_Mirrormaker>
             }
           }
         }' \
     mdb.api.cloud.yandex.net:443 \
     yandex.cloud.mdb.kafka.v1.ConnectorService.Create
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге.

4. Чтобы создать коннектор S3 Sink, воспользуйтесь вызовом ConnectorService/Create и выполните запрос, например, с помощью gRPCurl:

   grpcurl \
     -format json \
     -import-path ~/cloudapi/ \
     -import-path ~/cloudapi/third_party/googleapis/ \
     -proto ~/cloudapi/yandex/cloud/mdb/kafka/v1/connector_service.proto \
     -rpc-header "Authorization: Bearer $IAM_TOKEN" \
     -d '{
           "cluster_id": "<идентификатор_кластера>",
           "connector_spec": {
             "name": "<имя_коннектора>",
             "tasks_max": {
               "value": "<лимит_задач>"
             },
             "properties": "<дополнительные_свойства_коннектора>"
             "connector_config_s3_sink": {
               <настройки_коннектора_S3-Sink>
             }
           }
         }' \
     mdb.api.cloud.yandex.net:443 \
     yandex.cloud.mdb.kafka.v1.ConnectorService.Create
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге.

5. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

• --cluster-name — имя кластера.

• --direction — направление коннектора:

  • ingress — если кластер является приемником.
  • egress — если кластер является источником.

• --tasks-max — количество одновременно работающих процессов. Рекомендуется указывать не менее 2 для равномерного распределения нагрузки репликации.

• --properties — список дополнительных настроек коннектора в формате <ключ>:<значение>, разделенный запятыми. Примеры ключей:

  • key.converter
  • value.converter

  Список общих настроек коннекторов см. в документации Apache Kafka®.

• --replication-factor — количество копий топика, хранящихся в кластере.

• --topics — шаблон для отбора реплицируемых топиков, имена топиков перечисляются через запятую или символ |. Можно использовать выражение .*, например analysis.*. Для переноса всех топиков укажите .*.

• --this-cluster-alias — префикс для обозначения этого кластера в настройках коннектора.

• --external-cluster — параметры внешнего кластера:

  • alias — префикс для обозначения внешнего кластера в настройках коннектора.

  • bootstrap-servers — список FQDN хостов-брокеров внешнего кластера с номерами портов для подключения, разделенный запятыми.

    Как получить FQDN хоста-брокера, см. в инструкции.

  • security-protocol — протокол подключения коннектора:

    • plaintext, sasl_plaintext – для подключений без SSL;
    • ssl, sasl_ssl – для подключений с SSL.

  • sasl-mechanism — механизм шифрования имени и пароля.

  • sasl-username — имя пользователя для подключения коннектора к внешнему кластеру.

  • sasl-password — пароль пользователя для подключения коннектора к внешнему кластеру.

  • ssl-truststore-certificates — список сертификатов в формате PEM.

• properties — список дополнительных настроек коннектора в формате <ключ>:<значение>, разделенный запятыми. Примеры ключей:

  • key.converter
  • value.converter

  Список общих настроек коннекторов см. в документации Apache Kafka®.

• topics — шаблон для отбора реплицируемых топиков, имена топиков перечисляются через запятую или символ |. Можно использовать выражение .*, например analysis.*. Для переноса всех топиков укажите .*.

• replication_factor — количество копий топика, хранящихся в кластере.

• source_cluster и target_cluster — параметры для подключения к кластеру-источнику и кластеру-приемнику:

  • alias — префикс для обозначения кластера в настройках коннектора.

    Примечание

    Топики в кластере-приемнике будут созданы с указанным префиксом.

  • this_cluster — опция для использования текущего кластера в качестве источника или приемника.

  • external_cluster — параметры для подключения к внешнему кластеру:

    • bootstrap_servers — список FQDN хостов-брокеров кластера с номерами портов для подключения, разделенный запятыми.

      Как получить FQDN хоста-брокера, см. в инструкции.

    • sasl_username — имя пользователя для подключения коннектора к кластеру.

    • sasl_password — пароль пользователя для подключения коннектора к кластеру.

    • sasl_mechanism — механизм шифрования имени и пароля.

    • security_protocol — протокол подключения коннектора:

      • PLAINTEXT, SASL_PLAINTEXT – для подключений без SSL;
      • SSL, SASL_SSL – для подключений с SSL.

    • ssl_truststore_certificates — содержимое PEM-сертификата.

Настройки коннектора MirrorMaker задаются в параметре connectorSpec.connectorConfigMirrormaker:

• sourceCluster и targetCluster — параметры для подключения к кластеру-источнику и кластеру-приемнику:

  • alias — префикс для обозначения кластера в настройках коннектора.

    Примечание

    Топики в кластере-приемнике будут созданы с указанным префиксом.

  • thisCluster — опция для использования текущего кластера в качестве источника или приемника.

  • externalCluster — параметры для подключения к внешнему кластеру:

    • bootstrapServers — список FQDN хостов-брокеров кластера с номерами портов для подключения, разделенный запятыми.

      Как получить FQDN хоста-брокера, см. в инструкции.

    • saslUsername — имя пользователя для подключения коннектора к кластеру.

    • saslPassword — пароль пользователя для подключения коннектора к кластеру.

    • saslMechanism — механизм шифрования имени и пароля.

    • securityProtocol — протокол подключения коннектора:

      • PLAINTEXT, SASL_PLAINTEXT – для подключений без SSL;
      • SSL, SASL_SSL – для подключений с SSL.

    • sslTruststoreCertificates — содержимое PEM-сертификата.

• topics — шаблон для отбора реплицируемых топиков, имена топиков перечисляются через запятую или символ |. Можно использовать выражение .*, например analysis.*. Для переноса всех топиков укажите .*.

• replicationFactor — количество копий топика, хранящихся в кластере.

Настройки коннектора MirrorMaker задаются в параметре connector_spec.connector_config_mirrormaker:

• source_cluster и target_cluster — параметры для подключения к кластеру-источнику и кластеру-приемнику:

  • alias — префикс для обозначения кластера в настройках коннектора.

    Примечание

    Топики в кластере-приемнике будут созданы с указанным префиксом.

  • this_cluster — опция для использования текущего кластера в качестве источника или приемника.

  • external_cluster — параметры для подключения к внешнему кластеру:

    • bootstrap_servers — список FQDN хостов-брокеров кластера с номерами портов для подключения, разделенный запятыми.

      Как получить FQDN хоста-брокера, см. в инструкции.

    • sasl_username — имя пользователя для подключения коннектора к кластеру.

    • sasl_password — пароль пользователя для подключения коннектора к кластеру.

    • sasl_mechanism — механизм шифрования имени и пароля.

    • security_protocol — протокол подключения коннектора:

      • PLAINTEXT, SASL_PLAINTEXT – для подключений без SSL;
      • SSL, SASL_SSL – для подключений с SSL.

    • ssl_truststore_certificates — содержимое PEM-сертификата.

• topics — шаблон для отбора реплицируемых топиков, имена топиков перечисляются через запятую или символ |. Можно использовать выражение .*, например analysis.*. Для переноса всех топиков укажите .*.

• replication_factor — количество копий топика, хранящихся в кластере. Передается в виде объекта с полем value.

• --cluster-name — имя кластера.

• --tasks-max — количество одновременно работающих процессов. Рекомендуется указывать не менее 2 для равномерного распределения нагрузки репликации.

• --properties — список дополнительных настроек коннектора в формате <ключ>:<значение>, разделенный запятыми. Примеры ключей:

  • key.converter
  • value.converter
  • value.converter.schemas.enable
  • format.output.type

  Список всех настроек коннектора см. в документации коннектора. Список общих настроек коннекторов см. в документации Apache Kafka®.

• --topics — шаблон для отбора реплицируемых топиков, имена топиков перечисляются через запятую или символ |. Можно использовать выражение .*, например analysis.*. Для переноса всех топиков укажите .*.

• --file-compression-type — кодек для сжатия сообщений. После создания кластера данный параметр нельзя изменить. Допустимые значения:

  • none (по умолчанию) — сжатие отсутствует;
  • gzip — кодек gzip;
  • snappy — кодек snappy;
  • zstd — кодек zstd.

• --file-max-records — максимальное количество записей, которое может быть записано в один файл, размещенный в S3-совместимом хранилище.

• --bucket-name — имя бакета в S3-совместимом хранилище, в который будет производиться запись.

• --storage-endpoint — эндпоинт для доступа к хранилищу (его необходимо узнать у провайдера хранилища). Пример: storage.yandexcloud.net.

• --region — регион, в котором находится бакет S3-совместимого хранилища. Значение по умолчанию — us-east-1. Список доступных регионов.

• --access-key-id, --secret-access-key — идентификатор и содержимое AWS-совместимого ключа.

• properties — список дополнительных настроек коннектора в формате <ключ>:<значение>, разделенный запятыми. Примеры ключей:

  • key.converter
  • value.converter
  • value.converter.schemas.enable
  • format.output.type

  Список всех настроек коннектора см. в документации коннектора. Список общих настроек коннекторов см. в документации Apache Kafka®.

• topics — шаблон для отбора реплицируемых топиков, имена топиков перечисляются через запятую или символ |. Можно использовать выражение .*, например analysis.*. Для переноса всех топиков укажите .*.

• file_compression_type — кодек для сжатия сообщений. После создания кластера данный параметр нельзя изменить. Допустимые значения:

  • none (по умолчанию) — сжатие отсутствует;
  • gzip — кодек gzip;
  • snappy — кодек snappy;
  • zstd — кодек zstd.

• file_max_records — максимальное количество записей, которое может быть записано в один файл, размещенный в S3-совместимом хранилище.

• s3_connection — параметры для подключения к S3-совместимому хранилищу:

  • bucket_name — имя бакета, в который будет производиться запись.

  • external_s3 — параметры для подключения к внешнему S3-совместимому хранилищу:

    • endpoint — эндпоинт для доступа к хранилищу (его необходимо узнать у провайдера хранилища). Пример: storage.yandexcloud.net.

    • region — регион, в котором находится бакет S3-совместимого хранилища. Значение по умолчанию — us-east-1. Список доступных регионов.

    • access_key_id, secret_access_key — идентификатор и содержимое AWS-совместимого ключа.

Настройки коннектора S3 Sink задаются в параметре connectorSpec.connectorConfigS3Sink:

• topics — шаблон для отбора реплицируемых топиков, имена топиков перечисляются через запятую или символ |. Можно использовать выражение .*, например analysis.*. Для переноса всех топиков укажите .*.

• fileCompressionType — кодек для сжатия сообщений. После создания кластера данный параметр нельзя изменить. Допустимые значения:

  • none (по умолчанию) — сжатие отсутствует;
  • gzip — кодек gzip;
  • snappy — кодек snappy;
  • zstd — кодек zstd.

• fileMaxRecords — максимальное количество записей, которое может быть записано в один файл, размещенный в S3-совместимом хранилище.

• s3Connection — параметры для подключения к S3-совместимому хранилищу:

  • bucketName — имя бакета, в который будет производиться запись.
  • externalS3 — параметры внешнего хранилища:

    • endpoint — эндпоинт для доступа к хранилищу (его необходимо узнать у провайдера хранилища). Пример: storage.yandexcloud.net.

    • region — регион, в котором находится бакет S3-совместимого хранилища. Значение по умолчанию — us-east-1. Список доступных регионов.

    • accessKeyId, secretAccessKey — идентификатор и содержимое AWS-совместимого ключа.

Настройки коннектора S3 Sink задаются в параметре connector_spec.connector_config_s3_sink:

• topics — шаблон для отбора реплицируемых топиков, имена топиков перечисляются через запятую или символ |. Можно использовать выражение .*, например analysis.*. Для переноса всех топиков укажите .*.

• file_compression_type — кодек для сжатия сообщений. После создания кластера данный параметр нельзя изменить. Допустимые значения:

  • none (по умолчанию) — сжатие отсутствует;
  • gzip — кодек gzip;
  • snappy — кодек snappy;
  • zstd — кодек zstd.

• file_max_records — максимальное количество записей, которое может быть записано в один файл, размещенный в S3-совместимом хранилище. Передается в виде объекта с полем value.

• s3_connection — параметры для подключения к S3-совместимому хранилищу:

  • bucket_name — имя бакета, в который будет производиться запись.
  • external_s3 — параметры внешнего хранилища:

    • endpoint — эндпоинт для доступа к хранилищу (его необходимо узнать у провайдера хранилища). Пример: storage.yandexcloud.net.

    • region — регион, в котором находится бакет S3-совместимого хранилища. Значение по умолчанию — us-east-1. Список доступных регионов.

    • access_key_id, secret_access_key — идентификатор и содержимое AWS-совместимого ключа.

Если у вас еще нет интерфейса командной строки Yandex Cloud (CLI), установите и инициализируйте его.

По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

Чтобы изменить коннектор MirrorMaker:

1. Посмотрите описание команды CLI для изменения коннектора:

   yc managed-kafka connector-mirrormaker update --help
   

2. Запустите операцию, например, изменения лимита задач:

   yc managed-kafka connector-mirrormaker update <имя_коннектора> \
      --cluster-name=<имя_кластера> \
      --direction=<направление_коннектора> \
      --tasks-max=<новый_лимит_задач>
   

   Где --direction — направление коннектора: ingress или egres.

   Имя коннектора можно запросить со списком коннекторов в кластере, имя кластера — со списком кластеров в каталоге.

Чтобы изменить коннектор S3 Sink:

1. Посмотрите описание команды CLI для изменения коннектора:

   yc managed-kafka connector-s3-sink update --help
   

2. Запустите операцию, например, изменения лимита задач:

   yc managed-kafka connector-s3-sink update <имя_коннектора> \
      --cluster-name=<имя_кластера> \
      --tasks-max=<новый_лимит_задач>
   

   Имя коннектора можно запросить со списком коннекторов в кластере, имя кластера — со списком кластеров в каталоге.

1. Ознакомьтесь со списком настроек коннекторов MirrorMaker и S3 Sink.

2. Откройте актуальный конфигурационный файл Terraform с планом инфраструктуры.

   О том, как создать такой файл, см. в разделе Создание кластера.

3. Измените значение параметров в описании ресурса yandex_mdb_kafka_connector:

   • Для коннектора MirrorMaker:

     resource "yandex_mdb_kafka_connector" "<имя_коннектора>" {
       cluster_id = "<идентификатор_кластера>"
       name       = "<имя_коннектора>"
       tasks_max  = <лимит_задач>
       properties = {
         <дополнительные_свойства>
       }
       connector_config_mirrormaker {
         topics             = "<шаблон_для_топиков>"
         replication_factor = <фактор_репликации>
         source_cluster {
           alias = "<префикс_для_обозначения_кластера>"
           external_cluster {
             bootstrap_servers           = "<список_FQDN_хостов-брокеров>"
             sasl_username               = "<имя_пользователя>"
             sasl_password               = "<пароль_пользователя>"
             sasl_mechanism              = "<механизм_шифрования>"
             security_protocol           = "<протокол_безопасности>"
             ssl-truststore-certificates = "<содержимое_PEM-сертификата>"
           }
         }
         target_cluster {
           alias = "<префикс_для_обозначения_кластера>"
           this_cluster {}
         }
       }
     }
     

   • Для коннектора S3 Sink:

     resource "yandex_mdb_kafka_connector" "<имя_S3_Sink_коннектора>" {
       cluster_id = "<идентификатор_кластера>"
       name       = "<имя_S3_Sink_коннектора>"
       tasks_max  = <лимит_задач>
       properties = {
         <дополнительные_свойства>
      }
       connector_config_s3_sink {
         topics                = "<шаблон_для_топиков>"
         file_max_records      = <максимальное_количество_сообщений_в_файле>
         s3_connection {
           bucket_name = "<имя_бакета>"
           external_s3 {
             endpoint          = "<эндпоинт_S3-совместимого_хранилища>"
             access_key_id     = "<идентификатор_AWS-совместимого_статического_ключа>"
             secret_access_key = "<содержимое_AWS-совместимого_статического_ключа>"
           }
         }
       }
     }
     

4. Проверьте корректность настроек.

   1. В командной строке перейдите в каталог, в котором расположены актуальные конфигурационные файлы Terraform с планом инфраструктуры.

   2. Выполните команду:

      terraform validate
      

      Если в файлах конфигурации есть ошибки, Terraform на них укажет.

5. Подтвердите изменение ресурсов.

   1. Выполните команду для просмотра планируемых изменений:

      terraform plan
      

      Если конфигурации ресурсов описаны верно, в терминале отобразится список изменяемых ресурсов и их параметров. Это проверочный этап: ресурсы не будут изменены.

   2. Если вас устраивают планируемые изменения, внесите их:

      1. Выполните команду:

         terraform apply
         

      2. Подтвердите изменение ресурсов.

      3. Дождитесь завершения операции.

Подробнее см. в документации провайдера Terraform.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Воспользуйтесь методом Connector.update и выполните запрос, например, с помощью cURL:

   Важно

   Метод API переопределит все параметры изменяемого объекта, которые не были явно переданы в запросе, на значения по умолчанию. Чтобы избежать этого, перечислите настройки, которые вы хотите изменить, в параметре updateMask (одной строкой через запятую).

   curl \
     --request PATCH \
     --header "Authorization: Bearer $IAM_TOKEN" \
     --url 'https://mdb.api.cloud.yandex.net/managed-kafka/v1/clusters/<идентификатор_кластера>/connectors/<имя_коннектора>' \
     --data '{
               "updateMask": "connectorSpec.tasksMax,connectorSpec.properties,connectorSpec.connectorConfigMirrormaker.<настройка_коннектора_Mirrormaker_1>,...,connectorSpec.connectorConfigMirrormaker.<настройка_коннектора_Mirrormaker_N>,connectorSpec.connectorConfigS3Sink.<настройка_коннектора_S3-Sink_1>,...,connectorSpec.connectorConfigS3Sink.<настройка_коннектора_S3-Sink_N>",
               "connectorSpec": {
                 "tasksMax": "<лимит_задач>"
                 "properties": "<дополнительные_свойства_коннектора>"
                 "connectorConfigMirrormaker": {
                   <настройки_коннектора_Mirrormaker>
                 },
                 "connectorConfigS3Sink": {
                   <настройки_коннектора_S3-Sink>
                 }
               }
             }'
   

   Где:

   • updateMask — перечень изменяемых параметров коннектора в одну строку через запятую.

     Укажите нужные параметры:

     • connectorSpec.tasksMax – если нужно изменить лимит задач коннектора.
     • connectorSpec.properties – если нужно изменить дополнительные свойства коннектора.
     • connectorSpec.connectorConfigMirrormaker.<настройка_конфигурации_коннектора_Mirrormaker> – если нужно изменить настройки коннектора Mirrormaker.
     • connectorSpec.connectorConfigS3Sink.<настройка_конфигурации_коннектора_S3-Sink> – если нужно изменить настройки коннектора S3 Sink.

   • connectorSpec – укажите настройки коннектора MirrorMaker или S3 Sink.

   Идентификатор кластера можно запросить со списком кластеров в каталоге, а имя коннектора — со списком коннекторов в кластере.

3. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Клонируйте репозиторий cloudapi:

   cd ~/ && git clone --depth=1 https://github.com/yandex-cloud/cloudapi
   

   Далее предполагается, что содержимое репозитория находится в директории ~/cloudapi/.

3. Воспользуйтесь вызовом ConnectorService/Update и выполните запрос, например, с помощью gRPCurl:

   Важно

   Метод API переопределит все параметры изменяемого объекта, которые не были явно переданы в запросе, на значения по умолчанию. Чтобы избежать этого, перечислите настройки, которые вы хотите изменить, в параметре update_mask (в виде массива строк paths[]).

   Формат перечисления настроек

   "update_mask": {
       "paths": [
           "<настройка_1>",
           "<настройка_2>",
           ...
           "<настройка_N>"
       ]
   }
   

   grpcurl \
     -format json \
     -import-path ~/cloudapi/ \
     -import-path ~/cloudapi/third_party/googleapis/ \
     -proto ~/cloudapi/yandex/cloud/mdb/kafka/v1/connector_service.proto \
     -rpc-header "Authorization: Bearer $IAM_TOKEN" \
     -d '{
           "cluster_id": "<идентификатор_кластера>",
           "connector_name": "<имя_коннектора>",
           "update_mask": {
             "paths": [
               "connector_spec.tasks_max",
               "connector_spec.properties",
               "connector_spec.connector_config_mirrormaker.<настройка_коннектора_Mirrormaker_1>",
               ...,
               "connector_spec.connector_config_mirrormaker.<настройка_коннектора_Mirrormaker_N>",
               "connector_spec.connector_config_s3_sink.<настройка_коннектора_S3-Sink_1>",
               ...,
               "connector_spec.connector_config_s3_sink.<настройка_коннектора_S3-Sink_N>"
             ]
           },
           "connector_spec": {
             "tasks_max": {
               "value": "<лимит_задач>"
             },
             "properties": "<дополнительные_свойства_коннектора>"
             "connector_config_mirrormaker": {
               <настройки_коннектора_Mirrormaker>
             },
             "connector_config_s3_sink": {
               <настройки_коннектора_S3-Sink>
             }
           }
         }' \
     mdb.api.cloud.yandex.net:443 \
     yandex.cloud.mdb.kafka.v1.ConnectorService.Update
   

   Где:

   • update_mask — перечень изменяемых параметров коннектора в виде массива строк paths[].

     Укажите нужные параметры:

     • connector_spec.tasks_max – если нужно изменить лимит задач коннектора.
     • connector_spec.properties – если нужно изменить дополнительные свойства коннектора.
     • connector_spec.connector_config_mirrormaker.<настройка_конфигурации_коннектора_Mirrormaker> – если нужно изменить настройки коннектора Mirrormaker.
     • connector_spec.connector_config_s3_sink.<настройка_конфигурации_коннектора_S3-Sink> – если нужно изменить настройки коннектора S3 Sink.

   • connector_spec – укажите настройки коннектора MirrorMaker или S3 Sink.

   Идентификатор кластера можно запросить со списком кластеров в каталоге, а имя коннектора — со списком коннекторов в кластере.

4. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

Если у вас еще нет интерфейса командной строки Yandex Cloud (CLI), установите и инициализируйте его.

По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

Чтобы приостановить работу коннектора, выполните команду:

yc managed-kafka connector pause <имя_коннектора> \
   --cluster-name=<имя_кластера>

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Воспользуйтесь методом Connector.pause и выполните запрос, например, с помощью cURL:

   curl \
     --request POST \
     --header "Authorization: Bearer $IAM_TOKEN" \
     --url 'https://mdb.api.cloud.yandex.net/managed-kafka/v1/clusters/<идентификатор_кластера>/connectors/pause/<имя_коннектора>'
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге, а имя коннектора — со списком коннекторов в кластере.

3. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Клонируйте репозиторий cloudapi:

   cd ~/ && git clone --depth=1 https://github.com/yandex-cloud/cloudapi
   

   Далее предполагается, что содержимое репозитория находится в директории ~/cloudapi/.

3. Воспользуйтесь вызовом ConnectorService/Pause и выполните запрос, например, с помощью gRPCurl:

   grpcurl \
     -format json \
     -import-path ~/cloudapi/ \
     -import-path ~/cloudapi/third_party/googleapis/ \
     -proto ~/cloudapi/yandex/cloud/mdb/kafka/v1/connector_service.proto \
     -rpc-header "Authorization: Bearer $IAM_TOKEN" \
     -d '{
           "cluster_id": "<идентификатор_кластера>",
           "connector_name": "<имя_коннектора>"
         }' \
     mdb.api.cloud.yandex.net:443 \
     yandex.cloud.mdb.kafka.v1.ConnectorService.Pause
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге, а имя коннектора — со списком коннекторов в кластере.

4. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

Если у вас еще нет интерфейса командной строки Yandex Cloud (CLI), установите и инициализируйте его.

По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

Чтобы возобновить работу коннектора, выполните команду:

yc managed-kafka connector resume <имя_коннектора> \
   --cluster-name=<имя_кластера>

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Воспользуйтесь методом Connector.pause и выполните запрос, например, с помощью cURL:

   curl \
     --request POST \
     --header "Authorization: Bearer $IAM_TOKEN" \
     --url 'https://mdb.api.cloud.yandex.net/managed-kafka/v1/clusters/<идентификатор_кластера>/connectors/resume/<имя_коннектора>'
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге, а имя коннектора — со списком коннекторов в кластере.

3. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Клонируйте репозиторий cloudapi:

   cd ~/ && git clone --depth=1 https://github.com/yandex-cloud/cloudapi
   

   Далее предполагается, что содержимое репозитория находится в директории ~/cloudapi/.

3. Воспользуйтесь вызовом ConnectorService/Resume и выполните запрос, например, с помощью gRPCurl:

   grpcurl \
     -format json \
     -import-path ~/cloudapi/ \
     -import-path ~/cloudapi/third_party/googleapis/ \
     -proto ~/cloudapi/yandex/cloud/mdb/kafka/v1/connector_service.proto \
     -rpc-header "Authorization: Bearer $IAM_TOKEN" \
     -d '{
           "cluster_id": "<идентификатор_кластера>",
           "connector_name": "<имя_коннектора>"
         }' \
     mdb.api.cloud.yandex.net:443 \
     yandex.cloud.mdb.kafka.v1.ConnectorService.Resume
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге, а имя коннектора — со списком коннекторов в кластере.

4. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

Если у вас еще нет интерфейса командной строки Yandex Cloud (CLI), установите и инициализируйте его.

По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

Чтобы удалить коннектор, выполните команду:

yc managed-kafka connector delete <имя_коннектора> \
   --cluster-name <имя_кластера>

1. Откройте актуальный конфигурационный файл Terraform с планом инфраструктуры.

   О том, как создать такой файл, см. в разделе Создание кластера.

2. Удалите ресурс yandex_mdb_kafka_connector с описанием нужного коннектора.

3. Проверьте корректность настроек.

   1. В командной строке перейдите в каталог, в котором расположены актуальные конфигурационные файлы Terraform с планом инфраструктуры.

   2. Выполните команду:

      terraform validate
      

      Если в файлах конфигурации есть ошибки, Terraform на них укажет.

4. Подтвердите изменение ресурсов.

   1. Выполните команду для просмотра планируемых изменений:

      terraform plan
      

      Если конфигурации ресурсов описаны верно, в терминале отобразится список изменяемых ресурсов и их параметров. Это проверочный этап: ресурсы не будут изменены.

   2. Если вас устраивают планируемые изменения, внесите их:

      1. Выполните команду:

         terraform apply
         

      2. Подтвердите изменение ресурсов.

      3. Дождитесь завершения операции.

Подробнее см. в документации провайдера Terraform.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Воспользуйтесь методом Connector.pause и выполните запрос, например, с помощью cURL:

   curl \
     --request DELETE \
     --header "Authorization: Bearer $IAM_TOKEN" \
     --url 'https://mdb.api.cloud.yandex.net/managed-kafka/v1/clusters/<идентификатор_кластера>/connectors/<имя_коннектора>'
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге, а имя коннектора — со списком коннекторов в кластере.

3. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.

1. Получите IAM-токен для аутентификации в API и поместите токен в переменную среды окружения:

   export IAM_TOKEN="<IAM-токен>"
   

2. Клонируйте репозиторий cloudapi:

   cd ~/ && git clone --depth=1 https://github.com/yandex-cloud/cloudapi
   

   Далее предполагается, что содержимое репозитория находится в директории ~/cloudapi/.

3. Воспользуйтесь вызовом ConnectorService/Delete и выполните запрос, например, с помощью gRPCurl:

   grpcurl \
     -format json \
     -import-path ~/cloudapi/ \
     -import-path ~/cloudapi/third_party/googleapis/ \
     -proto ~/cloudapi/yandex/cloud/mdb/kafka/v1/connector_service.proto \
     -rpc-header "Authorization: Bearer $IAM_TOKEN" \
     -d '{
           "cluster_id": "<идентификатор_кластера>",
           "connector_name": "<имя_коннектора>"
         }' \
     mdb.api.cloud.yandex.net:443 \
     yandex.cloud.mdb.kafka.v1.ConnectorService.Delete
   

   Идентификатор кластера можно запросить со списком кластеров в каталоге, а имя коннектора — со списком коннекторов в кластере.

4. Убедитесь, что запрос был выполнен успешно, изучив ответ сервера.