{ "version": "https:\/\/jsonfeed.org\/version\/1.1", "title": "Slavlotski: заметки с тегом Kafka", "_rss_description": "Меня зовут Влад и я Data Engineer. В свободное от работы время люблю писать электронную музыку, играть в видеоигры", "_rss_language": "ru", "_itunes_email": "", "_itunes_categories_xml": "", "_itunes_image": "", "_itunes_explicit": "", "home_page_url": "https:\/\/slavlotski.com\/tags\/kafka\/", "feed_url": "https:\/\/slavlotski.com\/tags\/kafka\/json\/", "icon": "https:\/\/slavlotski.com\/pictures\/userpic\/userpic@2x.jpg?1696305806", "authors": [ { "name": "Slavlotski", "url": "https:\/\/slavlotski.com\/", "avatar": "https:\/\/slavlotski.com\/pictures\/userpic\/userpic@2x.jpg?1696305806" } ], "items": [ { "id": "15", "url": "https:\/\/slavlotski.com\/all\/apache-kafka\/", "title": "Apache Kafka", "content_html": "
\n\"\"\n
Источник: pinterest.com<\/a><\/div>\n<\/div>\n

Message broker<\/b> — это тип построения архитектуры, при котором элементы системы «общаются» друг с другом с помощью посредника. Данная архитектура нужна чтобы доставлять сообщения из пункта А в пункт Б, причем мы предполагаем, что эти сообщения бесконечны. Таким образом потоковая передача данных отличается от пакетной (пакетная рано или поздно завершится, имеет границы и ее можно разделить на эти границы). Message broker’ы активно используются в микросервисной архитектуре, где используется событийно-ориентированный подход. Преимуществами микросервиса над монолитным приложением являются низкая связь сервисов друг с другом, устойчивость приложений к сбоям за счет изоляции поставщиков (producer) и потребителей (consumer).<\/p>\n

Типы Message broker’ов:
\n— point-to-point<\/b>, брокеры которые работают на принципе доставки сообщения в строгой последовательности в виде очереди, где одна система пишет сообщение по принципу first in\/ first out, другая очередь эти сообщения вычитывает. Примеры таких брокеров: ZeroMQ, nanomsg, Java Message service (JMS)
\n— publish \/ subscribe<\/b> Есть некий producer, который публикует свои сообщения, есть так называемые потребители (consumer), которые эти сообщения получают именно по подписке. Строгая последовательность доставки сообщений не гарантируется. Системы с таким типом являются более масштабируемыми. Примеры таких брокеров: Celery, ActiveMQ, Apache Kafka, IBM MQ<\/p>\n

В терминалогии потоковых данных событие генерируется 1 раз инициатором, а затем обрабатываются различными потребителями. Инициаторов может быть много, кто передает эти данные, также и потребителей может быть много.<\/p>\n

Назначение Message broker:
\n— интеграция систем, написанные на разных языках программирования и протоколах
\n— гарантия надежного хранения
\n— гарантированная доставка
\n— масштабирование (как источников, так и потребителей)
\n— преобразование сообщений<\/p>\n

Apache Kafka<\/b> — это распределенная и легко масштабируемая система обмена сообщениями, написанная на Java и Scala, с высокой пропускной способностью, которая может в режиме реального времени обрабатывать большие объемы данных. Kafka появилась из-за необходимости компании LinkedIn эффективно перемещать огромные количества сообщений — до нескольких терабайт в час.<\/p>\n

Верхнеуровнево Kafka состоит из:
\n— Broker
\n— ZooKeeper или KRaft
\n— Consumer
\n— Producer<\/p>\n

\n\"\"\n
Источник: habr.com<\/a><\/div>\n<\/div>\n

Broker<\/b> — это серверное программное обеспечение с доступом к своему локальному диску, в которое producer’ы записывают данные, а consumer’ы читают данные, Broker эти данные аккумулирует и правильно сохраняет. Apache Kafka кластер состоит из множества Broker’ов, которые объединены в одну сеть.<\/p>\n

ZooKeeper<\/b> — это распределенное файловой хранилище необходимое для достижения согласованного состояния и синхронизации Broker’ов. Благодаря ZooKeeper мы можем управлять кластером Kafka, добавлять новых пользователей, создавать топики, задавать им настройки, обнаруживать сбои и восстанавливать работу кластера, хранить конфигурацию и секреты, авторизационные данные и ограничения или Access Control Lists при работе консумеров и продюсеров с брокерами.<\/p>\n

Consumer<\/b> — это приложение, которое имеет модуль Kafka, с помощью которого оно может прочитать сообщение. Приложение-консумер подписывается на события и получает данные в ответ.<\/p>\n

Producer<\/b> — это приложение, которое имеет модуль Kafka, с помощью которого оно может записать событие (сообщение) в кластер Kafka. Кластер сохраняет эти события и возвращает подтверждение о записи или acknowledgement.<\/p>\n

Брокеры<\/h2>\n

Чтобы Broker’ы знали куда нужно отправить сообщение producer’а и какие consumer’ы могут читать эти сообщения существует такое понятие как Topic<\/b><\/p>\n

\n\"\"\n
Источник: lydtechconsulting.com<\/a><\/div>\n<\/div>\n

Topic<\/b> — это базовая, основная сущность Apache Kafka, логическое<\/b> разделение коллекции связанных сообщений на группы, последовательность событий. Топик удобно представить в виде лога, в который постоянно добавляются новые данные в конец, тем самым не разрушается цепочка старых событий. Отличие топика Kafka от остальных топиков очередей тем что данные в топике Kafka нельзя удалить используя Consumer или Producer.<\/p>\n

Topic состоит из партиций<\/b>, которых может быть одна или несколько. Партиции — это главный механизм масштабирования и отказоустойчивости. 1 партиция — это 1 копия данных. Партиция может находится как на одном брокере, так и на нескольких. Сколько нужно партиций зависит от того насколько много consumer’ов читает топик и насколько часто они читают из этого топика. Если довольно часто, то в идеале для каждого consumer’а иметь свою отдельную партицию.<\/p>\n

Сами партиции физически представлены на дисках в виде сегментов<\/b>. Сегмент — это отдельный файл, хранящийся на диске брокера. Файлы можно создавать, ротировать с одного брокера на другой, удалять согласно настройке устаревания. Число сегментов может варьироваться в зависимости от интенсивности записи и настроек размера сегмента. При превышении размера сегмента создается новый.<\/p>\n

Хранение данных<\/h2>\n

Семантически и физически сообщения внутри сегмента не могут быть удалены, они иммутабельны. Всё, что мы можем — указать, как долго Kafka-брокер будет хранить события через настройку политики устаревания данных retention policy<\/b> и указать cleanup policy<\/b>, когда наступает некое событие для очистки данных.<\/p>\n

Retention policy<\/b> правила, которые позволяют избавляться от устаревших данных на основании времени. При достижении порога данные помечаются на удаление. Существуют следующие опции настройки:
\n— по милисекундам log.retention.ms<\/b>
\n— по минутам log.retention.minutes<\/b>
\n— по часам log.retention.hours<\/b>
\nПомимо этого существует size based подход, где оценивается размер сегмента, который может хранить топик:
\n— log.segment.bytes<\/b><\/p>\n

Cleanup policy<\/b> состоит из:
\n— Delete (по умолчанию). Помечает на удаление segment при устаревании \/ превышении размера
\n— Compact. Оставляет только последние сообщения для каждого ключа (message key)
\n— Delete и Compact. Производится compaction и удаление согласно retention policy<\/p>\n

Репликация данных<\/h2>\n

Для обеспечения отказоустойчивости и сохранности данных существует механизм репликации между брокерами, который имплементируется на уровне партиций:
\n— У каждой партиции есть настраиваемое число реплик
\n— Одна из этих реплик называется партицией-лидером<\/b>, которая принимает все запросы на запись\/чтение данных. Все остальные являются партициями-фолловерами<\/b>
\n— Записанные данные в партицию лидера автоматически реплицируются фолловерами внутри кластера Kafka. Фолловеры подключаются к лидеру, читают данные и асихронно сохраняют к себе на диск
\n— Роли лидеров и фолловеров не статичны. В случае выхода из строя брокера с лидирующими партициями, роль лидера достанется одной из реплик фолловеров, а консумеры и продюсеры получат обновление о необходимости переподключиться к брокерам с новыми лидерами партиций
\n— Начиная с версии Kafka 2.4 консумеры могут читать с партиций фолловеров. Это полезно для сокращения задержек при обращении к ближайшему брокеру в одной зоне доступности. Однако, из-за асинхронной работы репликации, взамен вы получаете от фолловеров менее актуальные данные, чем они есть в лидерской партиции<\/p>\n

\n\"\"\n
Источник: habr.com<\/a><\/div>\n<\/div>\n

Что из себя представляет сообщение Kafka<\/h2>\n

Каждое событие — это пара ключ-значение. Ключ партицирования может быть любой: числовой, строковый, объект или вовсе пустой. Значение тоже может быть любым — числом, строкой или объектом в своей предметной области, который вы можете как-то сериализовать (JSON, Protobuf, …) и хранить. В сообщении продюсер может указать время, либо за него это сделает брокер в момент приёма сообщения. Заголовки выглядят как в HTTP-протоколе — это строковые пары ключ-значение. В них не следует хранить сами данные, но они могут использоваться для передачи метаданных.<\/p>\n

\n\"\"\n
Источник: habr.com<\/a><\/div>\n<\/div>\n

Запись данных producer’ом<\/h2>\n

Чтобы producer смог отправить данные в Kafka кластер ему необходимо знать IP адреса всех брокеров и название топика. Перед тем как producer пойдет записывать данные он опросит брокер из своего пула и уточнит какая его партиция является лидером в топике и какое местоположение этой партиции в файловой системе. Запись осуществляется только в партицию-лидер.<\/p>\n

Продюсер определяет стратегию партицирования, она может быть как по ключу сообщения<\/b>, по очереди<\/b> (round-robin), так и кастомная<\/b>, реализованная на стороне продюсера. По ключу сообщения одного и того же идентификатора сохраняются в одну партицию. Примером такого ключа может быть номер карты, ID клиента и т. д. Например, ключ со значением Prod_id_1 всегда будет сохраняться в партицию 0, а со значением Prod_id_2 будет сохраняться в партицию 1, то есть данные не будут распределены по всем имеющимся партициям.<\/p>\n

\n\"\"\n
Источник: javapoint<\/a><\/div>\n<\/div>\n

При round-robin сообщения попадают в партиции по очереди, такая стратегия хорошо работает когда нужно равномерно распределить сообщения между всеми существующими партициями и очередность не играет роли.<\/p>\n

\n\"\"\n
Источник: javapoint<\/a><\/div>\n<\/div>\n

Семантики доставки сообщений<\/b>
\nВ очередях есть выбор между скоростью доставки и расходами на надежность доставки.
\n— at-most once<\/b> — при доставке сообщений устраивают потери сообщений, но не их дубликаты. Это самая слабая гарантия, которую реализуют брокерами очередей
\n— at-least once<\/b> — не хотим терять сообщения, но нас устраивают дубликаты
\n— exactly-once<\/b> — хотим доставить одно и только одно сообщение ничего не теряя и ничего не дублируя. Высокая надёжность данной семантики означает большие задержки<\/p>\n

\n\"\"\n
Источник: habr.com<\/a><\/div>\n<\/div>\n

Надежность доставки<\/b>
\nНадежность доставки продюсером данных в брокер осуществляется с помощью возвращения подтверждения записи данных продюсеру в партиции регулируется параметром acks<\/b>.
\n— При значении acks=0<\/b> продюсер не получает подтверждение от брокера, что запись произошла успешно. Данная политика влечет за собой риски потери данных
\n— При значении acks=1<\/b> продюсер будет ожидать получения ответа об успешной записи данных от лидера партиции
\n— При значении acks=all<\/b> продюсер ожидает подтверждения и от лидера партиции, и от фолловеров партиции. Данный вариант является самым надежным и самым трудозатратным, так как требует больше накладных расходов: мало того, чтобы нужно сохранить на диск, так ещё и дождаться, пока фолловеры отреплицируют сообщения и сохранят их к себе на диск. При включенной опции enable.idempotence<\/b> сообщениям проставляется PID (идентификатор продюсера) и увеличивающийся sequence number. Таким образом обеспечивается транзакционность и в случае сбоя в сети при попытке доставить сообщение повторно сообщения с одинаковым PID будут отброшены со стороны брокера.<\/p>\n

Чтение данных consumer’ом<\/h2>\n

Консумеры читают данные синхронно или асинхронно из лидерской партиции — это позволяет достичь консистентности при работе с данными. Информация в партиции читается слева-направо. Консумеров, читающих сообщения из топика, может быть несколько, причем читать они могут с разных позиций партиции, тем самым не мешая друг другу. Также нет какой-то привязки к чтению по времени, в зависимости от задачи консумеры могут читать спустя дни, недели, месяцы или несколько раз через какое-то время. Сама Kafka (в данном случае брокер) не следит за тем какое сообщение будет читать consumer и когда ему приходить за этими сообщениями. Consumerы сами должны ходить в Kafka и читать оттуда сообщения, сами должны говорить Kafka какие сообщения им выдать<\/p>\n

\n\"\"\n
Источник: habr.com<\/a><\/div>\n<\/div>\n

Offset<\/b> — это позиция сообщения в очереди Kafka. Начальная позиция в сообщении называется log-start offset<\/b>. Позиция сообщения, записанного последним — log-end offset<\/b>. Позиция консумера сейчас — current offset<\/b>. Расстояние между конечным оффсетом и текущим оффсетом консумера называют лагом — это первое, за чем стоит следить в своих приложениях. Допустимый лаг для каждого приложения свой, это тесно связано с бизнес-логикой и требованиями к работе системы.<\/p>\n

Consumer при чтении умеет запоминать на каком offset он остановился. Сохранение события фиксации позиции сообщения происходит в специальный Kafka топик _consumer_offset<\/b>. Под событием фиксации понимается так называемые commits<\/b>, их может быть два:
\n— auto commit<\/b> (по умолчанию). Он вычитывает кол-во offset (батч оффсетов) и потом фиксирует, какое кол-во offsetов он прочитал.
\n— user commit<\/b>. Можно организовать свою логику фиксирования коммитов, например, прочитал 1 offset, зафиксировал коммит. Минус такого подхода — снижение производительности<\/p>\n

Как обеспечивается то, что консумеры читают разные данные, разные партиции, а не одни и теже:
\n— Для этого есть такое понятие как консумер группы (consumer groups)<\/b> — это множество консумеров объединившихся в один кластер.
\n— Каждый консумер в группе будет читать разные сообщения. Каждый консумер пойдет в свою партицию и будет читать именно ее. Читать будет с того места где он остановился прошлый раз
\n— Kafka сохраняет на своей стороне текущий оффсет по каждой партиции топиков, которые входят в состав консумер-группы. Консумер в группе, после обработки прочитанных сообщений отправляет запрос на сохранение оффсета — или коммитит свой оффсет
\n— Распределение партиций между консумерами в пределах одной группы выполняется автоматически на стороне брокеров. Kafka старается честно распределять партиции между консумер-группами, насколько это возможно.
\n— Консумер группы создаются для решения разных кейсов. То есть данные могут быть одни и те же, но пользователи и задачи решаемые этими пользователями будут разные. Например, один консумер использует логи авторизаций пользователей для нужд администраторов, а другая консумер группа в виде маркетологов нужно смотреть кол-во авторизовавшихся пользователей на странице.
\n— Каждая такая группа имеет свой идентификатор, что позволяет регистрироваться на брокерах Kafka. Пространство имён консумер-групп глобально, а значит их имена в кластере Kafka уникальны.<\/p>\n

\n\"\"\n
Источник: habr.com<\/a><\/div>\n<\/div>\n

Ребалансировка консумер-групп<\/b>
\nПри добавлении новых потребителей топика происходит ребалансировка консумер-группы. Процесс ребалансировки заставляет все консумеры прекратить чтение и дождаться полной синхронизации участников, чтобы обрести новые партиции для чтения. Как только группа стала стабильной, а её участники получили партиции, консумеры в ней начинают чтение. Поскольку группа новая и раньше не существовала, то консумер выбирает позицию чтения оффсета: с самого начала earliest или же с конца latest. Топик мог существовать несколько месяцев, а консумер появился совсем недавно. В таком случае важно решить: читать ли все сообщения или же достаточно читать с конца самые последние, пропустив историю. Выбор между двумя опциями зависит от бизнес-логики протекающих внутри топика событий.<\/p>\n

Для того чтобы понимать какие из участников группы активны и работают, а какие уже нет, каждый консумер группы в равные промежутки времени отправляет Heartbeat-сообщение<\/b>. Временное значение настраивается программой-консумером перед запуском. Также консумер объявляет время жизни сессии<\/b> — если за это время он не смог отправить ни одно из Heartbeat-сообщений брокеру, то покидает группу. Брокер, в свою очередь, не получив ни одно из Heartbeat-сообщений консумеров, запускает процесс ребалансировки консумеров в группе.<\/p>\n

Процесс ребалансировки проходит достаточно болезненно для больших консумер-групп с множеством топиков. Поэтому разработчикам программ-консумеров обычно рекомендуют использовать по одной консумер-группе на топик. Также полезно держать число потребителей не слишком большим, чтобы не запускать ребалансировку много раз, но и не слишком маленьким, чтобы сохранять производительность и надёжность при чтении. Значения интервала Heartbeat и время жизни сессии следует устанавливать так, чтобы Heartbeat-интервал был в три-четыре раза меньше session timeout. Сами значения нужно выбирать не слишком большими, чтобы не увеличивать время до обнаружения «выпавшего» консумера из группы, но и не слишком маленьким, чтобы в случае малейших сетевых проблем, группа не уходила в ребалансировку.<\/p>\n

\n\"\"\n
Источник: habr.com<\/a><\/div>\n<\/div>\n

Ещё один гипотетический сценарий: партиций в топике 4, а консумеров в группе 5. В этом случае группа будет стабилизирована, однако участники, которым не достанется ни одна из партиций, будут бездействовать. Такое происходит потому, что с одной партицией в группе может работать только один консумер, а два и более консумеров не могут читать из одной партиции в группе. Отсюда возникает следующая базовая рекомендация: устанавливайте достаточное число партиций на старте, чтобы вы могли горизонтально масштабировать ваше приложение.<\/p>\n

Бонус:<\/h2>\n

Неплохой ролик о том какие проблемы решает Kafka, какие его преимущества в сравнении с другими брокерами сообщений и многое другое. В ролике отличная анимация и демонстрация работы основных компонентов Kafka<\/p>\n

\n