Передача данных в эндпоинт-приемник Apache Kafka®

• Тип эндпоинта — kafka_target.

• security_groups — группы безопасности для сетевого трафика.

  Правила групп безопасности применяются к трансферу. Они позволяют открыть сетевой доступ с ВМ трансфера к кластеру. Подробнее см. в разделе Сеть в Yandex Data Transfer.

  Группы безопасности должны принадлежать той же сети, в которой размещен кластер.

  Примечание

  В Terraform сеть для групп безопасности задавать не нужно.

• connection.cluster_id — идентификатор кластера, к которому необходимо подключиться.

• auth — метод аутентификации. Используется при подключении к хостам-брокерам:

  • sasl — аутентификация с помощью SASL.
  • no_auth — без аутентификации.

Пример структуры конфигурационного файла:

resource "yandex_datatransfer_endpoint" "<имя_эндпоинта_в_Terraform>" {
  name = "<имя_эндпоинта>"
  settings {
    kafka_target {
      security_groups = ["<список_идентификаторов_групп_безопасности>"]
      connection {
        cluster_id = "<идентификатор_кластера>"
      }
      auth {
        <метод_аутентификации>
      }
      <настройки_топика>
      <настройки_сериализации>
    }
  }
}

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

• securityGroups — группы безопасности для сетевого трафика, правила которых применятся к ВМ и кластерам без изменения их настроек. Подробнее см. в разделе Сеть в Yandex Data Transfer.

• connection.clusterId — идентификатор кластера, к которому необходимо подключиться.

• auth — метод аутентификации. Используется при подключении к хостам-брокерам:

  • sasl — аутентификация с помощью SASL. Необходимо указать следующие параметры:

    • user — имя учетной записи, от имени которой сервис Data Transfer будет подключаться к топику.
    • password.raw — пароль учетной записи (в текстовом виде).
    • mechanism — механизм хеширования.

  • noAuth — без аутентификации.

• Тип эндпоинта — kafka_target.

• security_groups — группы безопасности для сетевого трафика.

  Правила групп безопасности применяются к трансферу. Они позволяют открыть сетевой доступ с ВМ трансфера к хостам-брокерам. Подробнее см. в разделе Сеть в Yandex Data Transfer.

  Группы безопасности должны принадлежать той же сети, что и подсеть subnet_id, если она указана.

  Примечание

  В Terraform сеть для групп безопасности задавать не нужно.

• connection.on_premise — параметры подключения к хостам-брокерам:

  • broker_urls — IP-адреса или FQDN хостов-брокеров.

    Если номер порта Apache Kafka® отличается от стандартного, укажите его через двоеточие после имени хоста:

    <IP-адрес_или_FQDN_хоста-брокера>:<номер_порта>
    

  • tls_mode — параметры шифрования передаваемых данных, если оно требуется, например для соответствия требованиям PCI DSS.

    • disabled — отключено.
    • enabled — включено.

      • ca_certificate — сертификат CA.

        Важно

        Если не добавить сертификат, трансфер может завершиться ошибкой.

  • subnet_id — идентификатор подсети, в которой находятся хосты-брокеры. Трансфер будет использовать эту подсеть для доступа к ним.

• auth — метод аутентификации. Используется при подключении к хостам-брокерам:

  • sasl — аутентификация с помощью SASL.
  • no_auth — без аутентификации.

Пример структуры конфигурационного файла:

resource "yandex_datatransfer_endpoint" "<имя_эндпоинта_в_Terraform>" {
  name = "<имя_эндпоинта>"
  settings {
    kafka_target {
      security_groups = ["<список_идентификаторов_групп_безопасности>"]
      connection {
        on_premise {
          broker_urls = ["<список_IP-адресов_или_FQDN_хостов-брокеров>"]
          subnet_id  = "<идентификатор_подсети_c_хостами-брокерами>"
        }
      }
      auth = {
        <метод_аутентификации>
      }
      <настройки_топика>
      <настройки_сериализации>
    }
  }
}

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

• securityGroups — группы безопасности для сетевого трафика, правила которых применятся к ВМ и кластерам без изменения их настроек. Подробнее см. в разделе Сеть в Yandex Data Transfer.

• connection.onPremise — параметры подключения к хостам-брокерам:

  • brokerUrls — IP-адреса или FQDN хостов-брокеров.

    Если номер порта Apache Kafka® отличается от стандартного, укажите его через двоеточие после имени хоста:

    <IP-адрес_или_FQDN_хоста-брокера>:<номер_порта>
    

  • tlsMode — параметры шифрования передаваемых данных, если оно требуется, например для соответствия требованиям PCI DSS.

    • disabled — отключено.
    • enabled — включено.

      • caCertificate — сертификат CA.

        Важно

        Если не добавить сертификат, трансфер может завершиться ошибкой.

  • subnetId — идентификатор подсети, в которой находятся хосты-брокеры. Трансфер будет использовать эту подсеть для доступа к ним.

• auth — метод аутентификации. Используется при подключении к хостам-брокерам:

  • sasl — аутентификация с помощью SASL. Необходимо указать следующие параметры:

    • user — имя учетной записи, от имени которой сервис Data Transfer будет подключаться к топику.
    • password.raw — пароль учетной записи (в текстовом виде).
    • mechanism — механизм хеширования.

  • noAuth — без аутентификации.

Укажите в блоке topic_settings одну из опций отправки сообщений в топик:

• topic — укажите параметры в этом блоке, чтобы отправлять все сообщения в один топик:

  • topic_name — имя топика, в который будут отправляться сообщения.
  • save_tx_order — опция, позволяющая сохранять порядок транзакций. Укажите значение true, чтобы не разбивать поток событий на независимые очереди по таблицам.

• topic_prefix — укажите префикс, чтобы отправлять сообщения в разные топики с заданным префиксом.

  Это аналог настройки Debezium database.server.name. Сообщения будут отправляться в топик с именем <префикс_топика>.<схема>.<имя_таблицы>.

Укажите в поле topicSettings одну из опций отправки сообщений в топик:

• topic — укажите параметры в этом поле, чтобы отправлять все сообщения в один топик:

  • topicName — имя топика, в который будут отправляться сообщения.
  • saveTxOrder — опция, позволяющая сохранять порядок транзакций. Укажите значение true, чтобы не разбивать поток событий на независимые очереди по таблицам.

• topicPrefix — укажите префикс, чтобы отправлять сообщения в разные топики с заданным префиксом.

  Это аналог настройки Debezium database.server.name. Сообщения будут отправляться в топик с именем <префикс_топика>.<схема>.<имя_таблицы>.

Укажите в блоке serializer выбранный тип сериализации:

• serializer_auto.

• serializer_json.

• serializer_debezium.

  Для этого типа можно указать параметры сериализации Debezium, задав их в блоке serializer_debezium.serializer_parameters в виде пар key/value.

Если вы хотите использовать JSON-схемы в Yandex Schema Registry, сохраняя совместимость схем при добавлении и удалении опциональных полей, добавьте в конфигурационный файл блок serializer с описанием настроек сериализации. Чтобы генерировать закрытую схему для ключей, добавьте в блок serializer переменную key.converter.dt.json.generate.closed.content.schema со значением true. Чтобы генерировать закрытую схему для значений, добавьте в блок serializer переменную value.converter.dt.json.generate.closed.content.schema со значением true.

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 = "<URL_пространства_имен_Schema_Registry>"
          }
          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.

  Важно

  В пространстве имен должна быть выставлена Политика проверок совместимости для JSON — optional friendly.

• значение_API-ключа — значение API-ключа с ограниченной областью действия для подключения к Schema Registry. Чтобы получить значение:

  1. Создайте API-ключ с ограниченной областью действия и поместите его в локальную переменную SECRET:

     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}'`
     

  2. Выведите в терминал значение переменной SECRET:

     echo $SECRET
     

  3. Скопируйте полученное значение и вставьте его в параметр value в конфигурационном файле.

Укажите в блоке serializer выбранный тип сериализации:

• serializerAuto.

• serializerJson.

• serializerDebezium.

  Для этого типа можно указать параметры сериализации Debezium, задав их в поле serializerDebezium.serializerParameters в виде пар key/value.

Если вы хотите использовать JSON-схемы в Yandex Schema Registry, сохраняя совместимость схем при добавлении и удалении опциональных полей, добавьте в тело запроса объект serializer с описанием настроек сериализации. Чтобы генерировать закрытую схему для ключей, добавьте в объект serializer переменную key.converter.dt.json.generate.closed.content.schema со значением true. Чтобы генерировать закрытую схему для значений, добавьте в объект serializer переменную value.converter.dt.json.generate.closed.content.schema со значением true.

"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": "<URL_пространства_имен_Schema_Registry>"
      },
      {
        "key": "value.converter.basic.auth.user.info",
        "value": "api-key:<значение_API-ключа>"
      }
    ]
  }
}

Где:

• URL_пространства_имен_Schema_Registry — эндпоинт пространства имен Schema Registry. Вы можете скопировать эндпоинт из подсказки по подключению к пространству имен Schema Registry на вкладке Debezium, в параметре value.converter.schema.registry.url.

  Важно

  В пространстве имен должна быть выставлена Политика проверок совместимости для JSON — optional friendly.

• значение_API-ключа — значение API-ключа с ограниченной областью действия для подключения к Schema Registry. Чтобы получить значение:

  1. Создайте API-ключ с ограниченной областью действия и поместите его в локальную переменную SECRET:

     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}'`
     

  2. Выведите в терминал значение переменной SECRET:

     echo $SECRET
     

  3. Скопируйте полученное значение и вставьте его в параметр value в конфигурационном файле.