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

1. Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

   Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.

2. Настройте доступ к кластеру-источнику из Yandex Cloud.

3. Настройте права доступа пользователя к нужному топику.

4. Выдайте права READ консьюмер-группе, идентификатор которой совпадает с идентификатором трансфера.

   bin/kafka-acls --bootstrap-server localhost:9092 \
     --command-config adminclient-configs.conf \
     --add \
     --allow-principal User:username \
     --operation Read \
     --group <идентификатор_трансфера>
   

5. (Опционально) Чтобы использовать авторизацию по логину и паролю, настройте SASL-аутентификацию.

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

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

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

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

  Примечание

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

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

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

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

• topic_names — список имен топиков, к которым необходимо подключиться.

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

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

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

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

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

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

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

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

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

• topicNames — список имен топиков, к которым необходимо подключиться.

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

• 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 — без аутентификации.

• topic_names — список имен топиков, к которым необходимо подключиться.

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

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.

• 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 — без аутентификации.

• topicNames — список имен топиков, к которым необходимо подключиться.

• transformer — правила трансформации.

• parser — правила конвертации, которые определяются одним из выбранных парсеров:

  • audit_trails_v1_parser — парсер для логов сервиса Audit Trails.

  • cloud_logging_parser — парсер для логов сервиса Cloud Logging.

  • 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 для этого атрибута, чтобы убрать кавычки из строковых переменных (если этого не сделать, значения строковых полей останутся без изменений).

• transformer — правила трансформации:

  • cloudFunction — идентификатор функции, созданной в сервисе Cloud Functions.

  • serviceAccountId — идентификатор сервисного аккаунта, от имени которого будет запускаться функция обработки.

  • numberOfRetries — количество попыток вызова функции обработки.

  • bufferSize — размер буфера (в байтах), при заполнении которого данные будут переданы функции обработки.

    Максимальный размер буфера — 3,5 МБ. Подробнее об ограничениях, действующих при работе с функциями в сервисе Cloud Functions, читайте в соответствующем разделе.

  • bufferFlushInterval — длительность интервала (в секундах), по истечении которого данные из потока должны быть переданы функции обработки.

    Примечание

    Если буфер заполнится или истечет интервал отправки, данные будут переданы функции обработки.

  • invocationTimeout — допустимое время ожидания ответа от функции обработки (в секундах).

  Важно

  Значения для bufferFlushInterval и invocationTimeout указываются с постфиксом s, например, 10s.

• parser — правила конвертации, которые определяются одним из выбранных парсеров:

  • auditTrailsV1Parser — парсер для логов сервиса Audit Trails.

  • cloudLoggingParser — парсер для логов сервиса Cloud Logging.

  • jsonParser или tskvParser — парсеры JSON и TSKV, соответственно.

    Поля, задающие параметры их работы, одинаковы для обоих парсеров:

    • dataSchema — схема данных, представленная либо в виде JSON-объекта с описанием схемы, либо в виде списка полей:

      • fields — схема в виде JSON-объекта. Объект содержит в себе массив JSON-объектов, которые описывают отдельные колонки.

      • jsonFields — схема в виде списка JSON-полей.

    • nullKeysAllowed — задайте значение true для этого поля, чтобы разрешить значение null в ключевых колонках.

    • addRestColumn — задайте значение true для этого поля, чтобы поля, отсутствующие в схеме, попадали в колонку _rest.

    • unescapeStringValues — задайте значение true для этого поля, чтобы убрать кавычки из строковых переменных (если этого не сделать, значения строковых полей останутся без изменений).