Skip to content

Брокер сообщений

Брокер сообщений - это компонент, используемый OpenLineage для хранения всех получаемых событий. Затем эти события обрабатываются консьюмером сообщений в пакетном режиме.

В настоящее время Data.Rentgen поддерживает только Apache Kafka в качестве брокера сообщений.

Почему Kafka?

Другие популярные реализации серверов OpenLineage используют HTTP протокол для получения событий. По нашему опыту, Kafka гораздо лучше подходит для этого случая:

  • Kafka разработана для масштабирования. Если производительности недостаточно, просто добавьте еще один брокер в кластер. Для HTTP серверов это не так просто, поскольку требует балансировки нагрузки на стороне обратного прокси или DNS.
  • Kafka разработана для получения МНОЖЕСТВА событий в секунду, например миллионов, и максимально быстрого сохранения их на диск. Поэтому никакие события не теряются, даже если консьюмером сообщений перегружен - события уже сохранены на диск и будут обработаны позже.
  • ETL скрипты в основном запускаются по расписанию. Обычный паттерн - почти никаких событий в течение дня, но огромные всплески каждый полный час (например, в 00:00, 01:00, 03:00, 12:00). Kafka используется как промежуточный буфер, который сглаживает эти всплески.
  • События, хранящиеся в Kafka, могут читаться пакетами, даже если интеграция OpenLineage изначально отправляет их по одному. Пакетная обработка дает в 10 раз лучшую производительность по сравнению с обработкой отдельных событий.
  • HTTP/HTTPS протокол имеет более высокую задержку, чем TCP протокол Kafka. Некоторые интеграции OpenLineage чувствительны к задержкам - например, документация слушателя заданий Flink явно говорит: Если вы заблокируете поток, вызывающий методы execute среды, возможно, будет заблокирован. Чем меньше времени требуется для отправки ответа, тем лучше.

Требования

  • Apache Kafka 3.x. Рекомендуется использовать последнюю версию Kafka.

Настройка

С Docker

  • Установите Docker

  • Установите docker-compose

  • Выполните следующую команду:

    $ docker compose --profile broker up -d --wait
...

    docker-compose загрузит образ Apache Kafka, создаст контейнер и том, а затем запустит контейнер.

    Точка входа образа создаст базу данных, если том пустой. Параметры можно задать через файл .env или секцию environment в docker-compose.yml

    docker-compose.yml

    broker: image: bitnamilegacy/kafka:3.9 restart: unless-stopped env_file: .env.docker ports: - 9093:9093 volumes: - kafka_data:/bitnami/kafka healthcheck: test: [CMD-SHELL, kafka-topics.sh --bootstrap-server 127.0.0.1:9095 --list] interval: 10s timeout: 5s retries: 5 profiles: - broker - consumer - http2kafka kafka_data:

    .env.docker

    Init Kafka

    KAFKA_CFG_NODE_ID=0 KAFKA_CFG_PROCESS_ROLES=controller,broker KAFKA_CFG_LISTENERS=DOCKER://:9092,LOCALHOST://:9093,CONTROLLER://:9094,INTERBROKER://:9095 KAFKA_CFG_ADVERTISED_LISTENERS=DOCKER://broker:9092,LOCALHOST://localhost:9093,INTERBROKER://broker:9095 KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,INTERBROKER:PLAINTEXT,DOCKER:SASL_PLAINTEXT,LOCALHOST:SASL_PLAINTEXT KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@broker:9094 KAFKA_CFG_INTER_BROKER_LISTENER_NAME=DOCKER KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER KAFKA_CFG_SASL_MECHANISM_CONTROLLER_PROTOCOL=PLAIN KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN KAFKA_CLIENT_USERS=data_rentgen KAFKA_CLIENT_PASSWORDS=changeme KAFKA_CFG_SASL_ENABLED_MECHANISMS=PLAIN,SCRAM-SHA-256

Без Docker

Пожалуйста, следуйте инструкции по установке Apache Kafka.