Подключиться
Практические руководства

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

Статья создана

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

  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 LTS из Cloud Marketplace и публичным 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. Подключиться к реестру схем и передать ему схемы форматов данных для ключа и значения.
    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-адрес, удалите его.

В этой статье: