Kafka — это хорошо известная платформа для потоковой передачи событий. В этой серии статей мы проведем базовое знакомство с Kafka (создание producer’ов и consumer’ов) и разберемся, как именно Kafka работает под капотом, чтобы вы могли лучше проектировать и настраивать ваши приложения.

Серия статей состоит из следующих связанных частей:

• Часть 1: Сообщение producer’а (текушая статья);

• Часть 2: Подъем consumer’ов;

• Часть 3: Смещения и как с ними справляться;

• Часть 4: Мой кластер исчез!!! — Охватываем неудачи.

Использующая простой конвейер данных для приема файлов, первая часть серии статей направлена на то, чтобы охватить следующее:

• Загрузку файлов данных (с данными о событиях переходов) в Kafka;

• Объяснение, как работает производящая сторона;

• Настройку конфигурации producer»а;

• Пропускная способность vs задержка.

Я буду использовать некоторые наборы данных для электронной коммерции, которые вы можете найти здесь. Примеры кода написаны на Kotlin, но реализацию должно быть легко перенести на любой язык программирования.

Вы можете найти исходный код на Github здесь.

Примечание: Если вы хотите запустить примеры кода, вам понадобится кластер, и вы можете найти файл docker‑compose в отчете.

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

Aiven для Kafka — это самый простой способ начать, и вы получаете множество дополнительных функций — например, возможность мониторинга (наблюдаемости) из коробки.

Давайте сразу погрузимся в суть.

Покажи мне код

Наш producer отправит некоторые события в Kafka. Модель данных для click стрима должна выглядеть аналогично следующей полезной нагрузке:

Я создам топик под названием ecommerce.events, который я буду использовать для хранения своих сообщений. Топик будет иметь 5 партиций и фактор репликации 3 (ведущий + 2 реплики).

Создавать продюсеры Kafka довольно просто, важной частью является создание и отправка записей.

Для каждого события мы создаем объект ProducerRecord и указываем топик, ключ (здесь разделяем на основе userId) и, наконец, полезную нагрузку события в качестве значения.

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

Если сообщение было успешно записано в Kafka, мы печатаем метаданные, в противном случае, если возвращается исключение, мы регистрируем его.

Но что на самом деле происходит, когда вызывается метод send()?

Kafka делает много вещей под капотом, когда вызывается метод send(), поэтому давайте опишем их ниже:

1. Сообщение сериализуется с помощью указанного сериализатора.

2. Специальная функция определяет, в какую партицию должно быть направлено сообщение.

3. Kafka внутри хранит буферы сообщений; у нас есть один буфер для каждой партиции, и каждый буфер может содержать множество пакетов сообщений, сгруппированных для каждой партиции.

4. Наконец, потоки I/O собирают эти пакеты и отправляют их брокерам.

   На данный момент наши сообщения находятся в процессе передачи от клиента к брокерам. Брокеры отправили/получили сетевые буферы для того, чтобы сетевые потоки смогли принимать сообщения и передавать их какому‑либо потоку I/O, чтобы фактически сохранить их на диске.

5. В leader broker сообщения записываются на диск и отправляются ведомым брокерам для репликации. Здесь следует отметить одну вещь: сообщения сначала записываются в кэш страниц и периодически сбрасываются на диск.

   (Примечание: кэширование страниц на диск — это крайний случай потери сообщений, но все же вы, возможно, захотите знать об этом)

6. Фолловеры (синхронизированные реплики) сохраняют и отправляют обратно подтверждение о том, что они воспроизвели сообщение.

7. Ответ RecordMetadata отправляется обратно клиенту.

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

9. Клиент получает ответ.

Компромиссы между задержкой и пропускной способностью

В мире распределенных систем большинство вещей связано с компромиссами, поэтому разработчик должен найти «золотую середину» между различными компромиссами; важно понимать, как все работает.

Одним из важных аспектов может быть настройка между пропускной способностью и задержкой. Некоторыми ключевыми конфигурациями для этого являются batch.size и linger.ms.

Маленький batch.size и установка linger на zero могут уменьшить время ожидания и помочь обрабатывать сообщения как можно скорее, но это может снизить пропускную способность. Настройка на низкую задержку также полезна в сценариях с низкой производительностью. Накопление меньшего количества записей, чем указанный batch.size, приведет к ожиданию клиентом linger.ms поступления новых записей.

С другой стороны, больший batch.size может потребовать больше памяти (поскольку мы выделим буферы для указанного размера пакета), но это увеличит пропускную способность. Здесь могут помочь другие параметры конфигурации, такие как compression.type, max.in.flight.requests.per.connection и max.request.size могут помочь здесь.

Давайте лучше проиллюстрируем это примером

Наши данные о событиях хранятся в CSV-файлах, которые мы хотим загрузить в Kafka, и поскольку, это не прием данных в реальном времени, мы на самом деле не заботимся о времени ожидания здесь, однако у нас хорошая пропускная способность, поэтому мы можем быстро их обрабатывать. При использовании конфигураций по умолчанию прием 1.000.000 сообщений занимает около 184 секунд.

Устанавливаем batch.size равным 64 КБ (по умолчанию 16), linger.ms больше 0 и, наконец, compression.type на gzip

Оказывает следующее влияние на время загрузки данных.

Примерно со 184 секунд снизился до 18 секунд. В обоих случаях ack=1.

Я оставлю за вами право поэкспериментировать и попробовать различные варианты конфигурации, чтобы увидеть, как они лучше сработают в зависимости от вашего варианта использования.

Наконец, если вас интересует смысл «только на 1 раз», установите enable.idempotency в значение true, что также приведет к тому, что для acks будет установлено значение all.

Подведение итогов

Мы посмотрели, как работает producer-сторона Kafka. В качестве рекомендаций при создании и развертывании приложений:

Подумайте о требованиях и попытайтесь настроить пропускную способность и время ожидания.

Apache Kafka для разработчиков, старт потока 10 марта.

Подумайте о гарантиях, которые вам должны предоставить ваши producer'ы; то есть семантическая идемпотентность “exactly once” и / или транзакции могут быть вашими друзьями.

Одна деталь, о которой ранее не упоминалось, но которую полезно знать, заключается в том, что если вы хотите создавать многопоточные приложения, лучше всего создать один экземпляр producer и разделить его между потоками.