Yandex Cloud
  • Сервисы
  • Решения
  • Почему Yandex Cloud
  • Сообщество
  • Тарифы
  • Документация
  • Связаться с нами
Подключиться
Language / Region
© 2022 ООО «Яндекс.Облако»
Yandex Managed Service for Apache Kafka®
  • Начало работы
  • Пошаговые инструкции
    • Все инструкции
    • Информация об имеющихся кластерах
    • Создание кластера
    • Подключение к кластеру
    • Остановка и запуск кластера
    • Обновление версии Apache Kafka®
    • Изменение настроек кластера
    • Управление хостами Apache Kafka®
    • Работа с топиками и разделами
    • Управление учетными записями Kafka
    • Управление коннекторами
    • Просмотр логов кластера
    • Удаление кластера
    • Мониторинг состояния кластера и хостов
  • Практические руководства
    • Все руководства
    • Настройка Kafka Connect для работы с Managed Service for Apache Kafka®
    • Использование схем формата данных с Managed Service for Apache Kafka®
      • Обзор
      • Работа с управляемым реестром схем формата данных
      • Работа с реестром схем формата данных Confluent
    • Миграция данных в Managed Service for Apache Kafka®
    • Поставка данных из Yandex Managed Service for PostgreSQL с помощью Debezium
    • Поставка данных из Yandex Managed Service for PostgreSQL с помощью Yandex Data Transfer
    • Поставка данных в Managed Service for ClickHouse
    • Поставка данных в ksqlDB
  • Концепции
    • Взаимосвязь ресурсов сервиса
    • Топики и разделы
    • Брокеры
    • Производители и потребители
    • Управление схемами данных
    • Классы хостов
    • Сеть в Managed Service for Apache Kafka®
    • Квоты и лимиты
    • Типы хранилища
    • Коннекторы
    • Техническое обслуживание
    • Настройки Apache Kafka®
  • Управление доступом
  • Правила тарификации
  • Справочник API
    • Аутентификация в API
    • gRPC (англ.)
      • Overview
      • ClusterService
      • ConnectorService
      • ResourcePresetService
      • TopicService
      • UserService
      • OperationService
    • REST (англ.)
      • Overview
      • Cluster
        • Overview
        • create
        • delete
        • get
        • list
        • listHosts
        • listLogs
        • listOperations
        • move
        • rescheduleMaintenance
        • start
        • stop
        • streamLogs
        • update
      • Connector
        • Overview
        • create
        • delete
        • get
        • list
        • pause
        • resume
        • update
      • ResourcePreset
        • Overview
        • get
        • list
      • Topic
        • Overview
        • create
        • delete
        • get
        • list
        • update
      • User
        • Overview
        • create
        • delete
        • get
        • grantPermission
        • list
        • revokePermission
        • update
      • Operation
        • Overview
        • get
  • История изменений
  • Вопросы и ответы
  1. Практические руководства
  2. Использование схем формата данных с Managed Service for Apache Kafka®
  3. Работа с реестром схем формата данных Confluent

Использование Confluent Schema Registry с Managed Service for Apache Kafka®

Статья создана
Yandex Cloud
  • Перед началом работы
  • Создайте топик для уведомлений об изменении схем форматов данных
  • Установите и настройте Confluent Schema Registry на виртуальной машине
  • Создайте скрипты производителя и потребителя
  • Проверьте правильность работы Confluent Schema Registry
  • Удалите созданные ресурсы

Apache Kafka® использует бинарный формат для хранения и передачи сообщений. В сообщениях нет никакой информации об их структуре. Поэтому для интерпретации данных в бинарном формате потребителю нужно воспользоваться схемой формата данных (format schema). Она описывает формат ввода или вывода данных.

На основе схемы формата данных производитель формирует, а потребитель интерпретирует сообщения из топиков. Если схемы форматов данных производителей и потребителей будут различаться, в работе приложения будут возникать ошибки из-за неправильной интерпретации сообщений.

Разработчикам производителя и потребителя необходимо:

  • Своевременное обновлять схемы форматов данных.
  • Обеспечивать возможность работы производителя и потребителя сразу с несколькими версиями схем форматов данных, если это необходимо.

Confluent Schema Registry — один из вариантов программного обеспечения, решающий проблему синхронизации схем форматов данных между производителями и потребителями.

Алгоритм его работы:

  1. Производитель передает схемы форматов данных в реестр схем Confluent Schema Registry в одном из поддерживаемых форматов:

    • Avro,
    • JSON Schema,
    • Protobuf.
  2. Confluent Schema Registry помещает схемы форматов данных на хранение в служебный топик Apache Kafka® с именем _schemas. При этом каждой схеме присваивается уникальный номер версии.

  3. Производитель указывает номер версии схемы в отправляемом сообщении.

  4. Потребитель получает сообщение и извлекает из него номер версии схемы формата данных.

  5. Если нужной схемы формата данных нет в локальном кеше, потребитель обращается за ней в реестр схем. Получив нужную схему, он корректно интерпретирует полученное сообщение.

Чтобы использовать Managed Service for Apache Kafka® совместно с Confluent Schema Registry:

  1. Создайте топик для уведомлений об изменении схем форматов данных.
  2. Установите и настройте Confluent Schema Registry на виртуальной машине.
  3. Создайте скрипты производителя и потребителя.
  4. Проверьте правильность работы Confluent Schema Registry.
  5. Удалите созданные ресурсы.

Перед началом работы

  1. Создайте кластер Managed Service for Apache Kafka® любой подходящей конфигурации.

    1. Создайте топик с именем messages для обмена сообщениями между производителем и потребителем.
    2. Создайте учетную запись с именем user и выдайте ей права на топик messages:
      • ACCESS_ROLE_CONSUMER,
      • ACCESS_ROLE_PRODUCER.
  2. В той же сети, что и кластер Managed Service for Apache Kafka®, создайте виртуальную машину c Ubuntu 20.04 и публичным IP-адресом.

  3. Чтобы разрешить прохождение трафика между кластером Managed Service for Apache Kafka® и виртуальной машиной, настройте группы безопасности.

  4. В группах безопасности виртуальной машины создайте правило для входящего трафика, разрешающее подключение через порт 8081 — через него производитель и потребитель будут обращаться к реестру схем:

    • Диапазон портов: 8081.
    • Протокол: TCP.
    • Назначение: CIDR.
    • CIDR блоки: 0.0.0.0/0 или диапазоны адресов подсетей, в которых работают производитель и потребитель.

Создайте топик для уведомлений об изменении схем форматов данных

  1. Создайте служебный топик с именем _schemas со следующими настройками:

    • Количество разделов: 1.
    • Политика очистки лога: Compact.

    Указанные значения настроек необходимы для работы Confluent Schema Registry.

  2. Создайте учетную запись с именем registry и выдайте ей права на топик _schemas:

    • ACCESS_ROLE_CONSUMER,
    • ACCESS_ROLE_PRODUCER.

    От имени этой учетной записи Confluent Schema Registry будет работать со служебным топиком _schemas.

Установите и настройте Confluent Schema Registry на виртуальной машине

  1. Подключитесь к виртуальной машине по SSH.

  2. Подключите репозиторий Confluent Schema Registry:

    wget -qO - https://packages.confluent.io/deb/6.2/archive.key | sudo apt-key add - && \
    sudo add-apt-repository "deb [arch=amd64] https://packages.confluent.io/deb/6.2 stable main"
    
  3. Установите пакеты:

    sudo apt-get update && \
    sudo apt-get install \
         confluent-schema-registry \
         openjdk-11-jre-headless \
         python3-pip --yes
    
  4. Получите SSL-сертификат.

  5. Создайте для сертификата защищенное хранилище:

    sudo keytool \
         -keystore /etc/schema-registry/client.truststore.jks \
         -alias CARoot \
         -import -file /usr/local/share/ca-certificates/Yandex/YandexCA.crt \
         -storepass <пароль защищенного хранилища сертификатов> \
         --noprompt
    
  6. Создайте файл /etc/schema-registry/jaas.conf с настройками для подключения к кластеру:

    KafkaClient {
      org.apache.kafka.common.security.scram.ScramLoginModule required
      username="registry"
      password="<пароль учетной записи registry>";
    };
    
  7. Измените файл /etc/schema-registry/schema-registry.properties, отвечающий за настройки Confluent Schema Registry:

    1. Закомментируйте строку:

      kafkastore.bootstrap.servers=PLAINTEXT://localhost:9092
      
    2. Раскомментируйте строку с параметром listeners. Она отвечает за сетевой адрес и порт, которые будет слушать Confluent Schema Registry. По умолчанию используется порт 8081 на всех сетевых интерфейсах:

      listeners=http://0.0.0.0:8081
      
    3. Добавьте в конец файла строки:

      kafkastore.bootstrap.servers=SASL_SSL://<FQDN 1-го хоста-брокера:9091>,<FQDN 2-го хоста-брокера:9091>,...,<FQDN N-го хоста-брокера:9091>
      kafkastore.ssl.truststore.location=/etc/schema-registry/client.truststore.jks
      kafkastore.ssl.truststore.password=<пароль защищенного хранилища сертификатов>
      kafkastore.sasl.mechanism=SCRAM-SHA-512
      kafkastore.security.protocol=SASL_SSL
      

      Список хостов-брокеров можно получить со списком хостов кластера.

  8. Измените файл с описанием модуля systemd /lib/systemd/system/confluent-schema-registry.service.

    1. Перейдите в блок [Service].

    2. Добавьте параметр Environment с настройками Java:

      ...
      
      [Service]
      Type=simple
      User=cp-schema-registry
      Group=confluent
      Environment="LOG_DIR=/var/log/confluent/schema-registry"
      Environment="_JAVA_OPTIONS='-Djava.security.auth.login.config=/etc/schema-registry/jaas.conf'"
      ...
      
  9. Обновите сведения о модулях systemd:

    sudo systemctl daemon-reload
    
  10. Запустите сервис Confluent Schema Registry:

    sudo systemctl start confluent-schema-registry.service
    
  11. Включите автоматический запуск Confluent Schema Registry после перезагрузки ОС:

    sudo systemctl enable confluent-schema-registry.service
    

Создайте скрипты производителя и потребителя

Приведенные скрипты отправляют и принимают сообщения в топике messages в виде пары ключ:значение. В качестве примера схемы форматов данных описаны в формате Avro.

Примечание

Скрипты на Python приведены в демонстрационных целях. Вы можете подготовить и передать схемы форматов данных и сами данные, создав аналогичный скрипт на другом языке программирования.

  1. Установите необходимые пакеты Python:

    sudo pip3 install avro confluent_kafka
    
  2. Создайте Python-скрипт потребителя.

    Алгоритм работы скрипта:

    1. Подключиться к топику messages и реестру схем Confluent Schema Registry.
    2. В непрерывном цикле считывать поступающие в топик messages сообщения.
    3. При получении сообщения запросить в реестре схем Confluent Schema Registry нужные схемы для разбора сообщения.
    4. Разобрать бинарные данные из сообщения в соответствии со схемами для ключа и значения и вывести результат на экран.

    consumer.py

    #!/usr/bin/python3
    
    from confluent_kafka.avro import AvroConsumer
    from confluent_kafka.avro.serializer import SerializerError
    
    
    c = AvroConsumer(
        {
            "bootstrap.servers": ','.join([
                "<FQDN 1-го хоста-брокера>:9091",
                ...
                "<FQDN N-го хоста-брокера>:9091",
            ]),
            "group.id": "avro-consumer",
            "security.protocol": "SASL_SSL",
            "ssl.ca.location": "/usr/local/share/ca-certificates/Yandex/YandexCA.crt",
            "sasl.mechanism": "SCRAM-SHA-512",
            "sasl.username": "user",
            "sasl.password": "<пароль учетной записи user>",
            "schema.registry.url": "http://<FQDN или IP-адрес сервера Confluent Schema Registry>:8081",
        }
    )
    
    c.subscribe(["messages"])
    
    while True:
        try:
            msg = c.poll(10)
    
        except SerializerError as e:
            print("Message deserialization failed for {}: {}".format(msg, e))
            break
    
        if msg is None:
            continue
    
        if msg.error():
            print("AvroConsumer error: {}".format(msg.error()))
            continue
    
        print(msg.value())
    
    c.close()
    
  3. Создайте Python-скрипт производителя.

    Алгоритм работы скрипта:

    1. Подключиться к реестру схем Confluent Schema Registry и передать ему схемы форматов данных для ключа и значения.
    2. Сформировать на основе переданных схем ключ и значение.
    3. Отправить в топик messages сообщение, состоящее из пары ключ:значение. Номера версий схем будут добавлены в сообщение автоматически.

    producer.py

    #!/usr/bin/python3
    
    from confluent_kafka import avro
    from confluent_kafka.avro import AvroProducer
    
    
    value_schema_str = """
    {
        "namespace": "my.test",
        "name": "value",
        "type": "record",
        "fields": [
            {
                "name": "name",
                "type": "string"
            }
        ]
    }
    """
    
    key_schema_str = """
    {
        "namespace": "my.test",
        "name": "key",
        "type": "record",
        "fields": [
            {
                "name": "name",
                "type": "string"
            }
        ]
    }
    """
    
    value_schema = avro.loads(value_schema_str)
    key_schema = avro.loads(key_schema_str)
    value = {"name": "Value"}
    key = {"name": "Key"}
    
    
    def delivery_report(err, msg):
        """Called once for each message produced to indicate delivery result.
        Triggered by poll() or flush()."""
        if err is not None:
            print("Message delivery failed: {}".format(err))
        else:
            print("Message delivered to {} [{}]".format(msg.topic(), msg.partition()))
    
    
    avroProducer = AvroProducer(
        {
            "bootstrap.servers": ','.join([
                "<FQDN 1-го хоста-брокера>:9091",
                ...
                "<FQDN N-го хоста-брокера>:9091",
            ]),
            "security.protocol": "SASL_SSL",
            "ssl.ca.location": "/usr/local/share/ca-certificates/Yandex/YandexCA.crt",
            "sasl.mechanism": "SCRAM-SHA-512",
            "sasl.username": "user",
            "sasl.password": "<пароль учетной записи user>",
            "on_delivery": delivery_report,
            "schema.registry.url": "http://<FQDN или IP-адрес сервера Schema Registry>:8081",
        },
        default_key_schema=key_schema,
        default_value_schema=value_schema,
    )
    
    avroProducer.produce(topic="messages", key=key, value=value)
    avroProducer.flush()
    

Проверьте правильность работы Confluent Schema Registry

  1. Запустите потребитель:

    python3 ./consumer.py
    
  2. В отдельном терминале запустите производитель:

    python3 ./producer.py
    
  3. Убедитесь, что данные, отправленные производителем, получены и правильно интерпретированы потребителем:

    {'name': 'Value'}
    

Удалите созданные ресурсы

Если вам больше не нужны созданные ресурсы, удалите виртуальную машину и кластер Managed Service for Apache Kafka®.

Если вы зарезервировали для созданной виртуальной машины публичный статический IP-адрес, удалите его.

Была ли статья полезна?

Language / Region
© 2022 ООО «Яндекс.Облако»
В этой статье:
  • Перед началом работы
  • Создайте топик для уведомлений об изменении схем форматов данных
  • Установите и настройте Confluent Schema Registry на виртуальной машине
  • Создайте скрипты производителя и потребителя
  • Проверьте правильность работы Confluent Schema Registry
  • Удалите созданные ресурсы