Kafka

Рассмотрим ключевые концепции Kafka, архитектурные особенности и наиболее распространенные сценарии использования.

Почему именно Kafka?

Вероятно, вы слышали о Kafka - это популярная распределенная платформа потоковой обработки событий с открытым исходным кодом. Согласно сайту проекта, ей доверяют около 80% компаний из списка Fortune 100. Если Kafka подходит крупным компаниям, значит, она вполне подойдет и на интервью по System Design.

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

В данной статье мы последовательно изучим Kafka: начнем с общего обзора и постепенно углубимся в детали архитектуры и внутренней реализации.

Практический пример: мотивация к изучению Kafka

Представьте себе чемпионат мира по футболу. Предположим, у вас есть веб-сайт, который предоставляет статистику матчей в реальном времени. Каждый раз, когда забивается гол, игрок получает предупреждение или производится замена, вам нужно обновить сайт.

Заведем сервер с очередью и будем отправлять в нее сообщения по мере поступления событий. Сервер или процесс, ответственный за отправку событий в очередь, называется производителем (producer). Далее у нас есть сервер, который считывает события из очереди и обновляет сайт. Этот сервер называют потребителем (consumer).

Теперь представьте, что чемпионат расширился до гипотетического турнира с участием тысяч команд, причем матчи проходят одновременно. Количество событий значительно увеличится, и ваш одиночный сервер, обслуживающий очередь, начнет испытывать трудности с обработкой этой нагрузки. Аналогично потребителю становится сложно справиться с таким объемом входящих данных.

Чтобы масштабировать эту систему, нам необходимо добавить больше серверов для распределения очереди. Но как убедиться, что события будут обрабатываться в правильном порядке?

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

Решение в том, чтобы распределять события на основе игры, которой они соответствуют. Таким образом, все события одной игры будут упорядочены, поскольку находятся в одной очереди. Это одна из ключевых идей Kafka: сообщения, отправляемые и принимаемые через Kafka, требуют стратегии распределения, заданной пользователем.

Но что делать с нашим потребителем, который продолжает испытывать перегрузку? Здесь в игру вступает концепция группы потребителей Kafka (consumer group). Каждое событие гарантированно обрабатывается лишь одним потребителем внутри группы.

Допустим, мы решили расширить нашу виртуальную версию чемпионата на другие виды спорта, такие как баскетбол. Однако мы хотим сохранить раздельное освещение спортивных событий: футбольному сайту не нужны баскетбольные новости, и наоборот. Для этого вводится понятие тем (topics). Каждая тема связана с определенной спортивной категорией и потребители будут подписываться на конкретную тему.

Таким образом, потребители событий игр футбола подпишутся исключительно на тему (topic) футбола, а потребители баскетбола - на тему (topic) баскетбола.

Основные понятия и архитектура Kafka

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

Брокеры (Brokers)

Кластер Kafka состоит из множества брокеров (broker) - отдельных серверов (физических или виртуальных). Каждый брокер хранит данные и обслуживает клиентов. Чем больше брокеров, тем больше данных можно хранить и тем большее количество клиентов можно обслуживать.

Разделы (Partitions)

Каждый брокер имеет ряд разделов. Раздел (partition) представляет собой упорядоченную неизменяемую последовательность сообщений, в которую сообщения только добавляются - своего рода файл журнала. Именно благодаря разделам Kafka масштабируется, позволяя организовать параллельную обработку сообщений.

Темы (Topics)

Тема это всего лишь логическое объединение разделов (partitions). Темы позволяют публиковать и получать доступ к данным в Kafka. Сообщения публикуются в одну из тем читаются из одной из тем. Темы всегда ориентированы на множество производителей (producers). Таким образом каждая тема может иметь ноль, одного или несколько производителей, записывающих данные.

Производители (Producers) и потребители (Consumers)

Производители (producers) - это процессы, записывающие данные в темы, а потребители (consumers) - процессы, читающие из них. Хотя Kafka предоставляет простой интерфейс для взаимодействия производителей и потребителей, создание и обработка самих сообщений ложится на разработчика. Kafka не заботится о содержании данных - она просто сохраняет и передает их.

Также заметим, что вы можете использовать Kafka либо как очередь сообщений, либо как систему обработки потоков (streams processing). Основное различие (не такое уж важное, по правде говоря) заключается в способе взаимодействия потребителя с данными:

• В очередях сообщений (message queues) потребитель считывает сообщения из очереди и подтверждает (acknowledges) обработку каждого элемента.

• В потоке данных (stream) потребитель непрерывно читает и обрабатывает сообщения по мере их поступления, без подтверждения завершения обработки. Это дает возможность реализовать более сложную обработку данных.

Как работает Kafka?

При возникновении события производитель формирует сообщение (или запись - record) и отправляет его в тему Kafka. Каждое сообщение включает обязательное значение (value) и три необязательных: ключ (key), временную метку (timestamp) и заголовки (headers).

Рассмотрим простой пример отправки сообщения в тему с использованием инструмента командной строки Kafka:

kafka-console-producer --broker-list localhost:9092 --topic my_topic --property
"parse.key=true" --property "key.separator=:"
> key1: Привет, Kafka с ключом!
> key2: Другое сообщение с другим ключом

Также посмотрим на аналогичный пример с использованием библиотеки kafkajs для Node.js:

// Инициализация клиента Kafka
const kafka = new Kafka({
  clientId: "my-app",
  brokers: ["localhost:9092"],
});
 
// Инициализация производителя
const producer = kafka.producer();
 
const run = async () => {
  // Подключение производителя
  await producer.connect();
 
  // Отправка сообщений в тему 'my_topic' с ключами
  await producer.send({
    topic: "my_topic",
    messages: [
      { key: "key1", value: "Привет, Kafka с ключом!" },
      { key: "key2", value: "Другое сообщение с другим ключом" },
    ],
  });
};

При публикации сообщения в тему Kafka сначала определяет подходящий раздел для него. Выбор раздела критически важен, так как влияет на распределение данных по кластеру. Процесс выбора раздела осуществляется следующим образом:

1. Определение раздела: Kafka применяет алгоритм хеширования ключа сообщения, назначая сообщение конкретному разделу. Если сообщение не имеет ключа, Kafka распределяет его используя алгоритм round-robin или руководствуется другим механизмом, указанным в конфигурации. Это гарантирует, что сообщения с одинаковым ключом всегда попадают в один и тот же раздел, сохраняя порядок на уровне раздела.

2. Назначение брокера: После выбора раздела Kafka определяет, какой брокер отвечает за этот раздел. Соответствие разделов конкретным брокерам управляется метаданными кластера Kafka. За этим следит контроллер Kafka (роль внутри кластера брокеров). Производитель использует эти метаданные для прямой отправки сообщения брокеру, содержащему целевой раздел.

Каждое сообщение в разделе Kafka представлено как запись в неизменяемом журнале (immutable commit log). Сообщения последовательно добавляются в конец журнала, поэтому Kafka часто описывают как распределенный журнал записей (distributed commit log). Такая структура журнала лежит в основе архитектуры Kafka и обладает несколькими важными преимуществами:

1. Неизменяемость: однажды записанные сообщения нельзя изменить или удалить. Эта особенность важна для производительности и надежности Kafka, упрощает репликацию, ускоряет восстановление и предотвращает проблемы согласованности, возникающие в системах, допускающих изменение данных.

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

3. Масштабируемость: простота механизма добавления записей в журнал способствует горизонтальному масштабированию. Можно добавлять новые разделы и распределять их по нескольким брокерам для увеличения пропускной способности. Каждый раздел может быть реплицирован на множестве брокеров для повышения отказоустойчивости.

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

С момента записи сообщений в указанный раздел, Kafka гарантирует сохранность и доступность данных посредством эффективной схемы репликации. Репликация основана на модели лидер-последователь (leader-follower):

1. Назначение лидера: каждому разделу назначается ведущий раздел (лидер, leader), расположенный на одном из брокеров. Ведущий раздел отвечает за обработку всех запросов на чтение и запись. Назначением ведущего раздела занимается контроллер кластера. Это нужно для равномерного распределения лидеров по всему кластеру для балансировки нагрузки.

2. Репликация последователей (followers): помимо ведущего раздела, существует несколько разделов-последователей (followers), расположенных на разных брокерах. Эти разделы пассивно копируют данные с ведущего, выступая резервными копиями на случай отказа последнего.

3. Синхронизация и согласованность: Последователи регулярно синхронизируются с ведущим разделом, чтобы поддерживать актуальность сообщений, записанных в журнал. Синхронизация крайне важна для поддержания согласованности данных в кластере. Если ведущий раздел выходит из строя, один из полностью синхронизированных разделов автоматически назначается новым лидером, сокращая время простоя и потери данных.

4. Контроллер Kafka координирует весь процесс репликации, следит за состоянием всех брокеров и управляет динамикой назначения лидеров и последователей. Когда брокер отключается, контроллер переназначает роль лидера одному из последователей, чтобы обеспечить непрерывную доступность раздела.

Наконец, потребители извлекают сообщения из тем Kafka, используя модель вытягивания (pull-based model). В отличие от некоторых других систем доставки сообщений, где данные принудительно отправляются потребителям, Kafka оставляет выбор потребителю, позволяя ему самостоятельно запрашивать новые сообщения через определенные интервалы. Такой подход имеет несколько преимуществ: контроль над скоростью потребления; упрощенное управление ошибками; предотвращение чрезмерной загрузки медленных потребителей; возможность эффективного пакетного чтения.

Для полноты картины приведем пример, как можно прочитать сообщения из темы my_topic с помощью консольной утилиты Kafka:

kafka-console-consumer --bootstrap-server localhost:9092 --topic my_topic
--from-beginning

# Результат
key1: Привет, Kafka с ключом!
key2: Другое сообщение с другим ключом

Аналогичный пример на JavaScript с библиотекой kafkajs выглядит так:

// Инициализация клиента Kafka
const kafka = new Kafka({
  clientId: "my-app",
  brokers: ["localhost:9092"],
});
 
// Инициализация потребителя
const consumer = kafka.consumer({ groupId: "my-group" });
 
const run = async () => {
  // Подключение потребителя
  await consumer.connect();
 
  // Подписка на тему 'my_topic'
  await consumer.subscribe({ topic: "my_topic" });
 
  // Чтение сообщений
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log({
        value: message.value.toString(),
        key: message.key.toString(),
      });
    },
  });
};

Объединив все вместе, получаем следующую картину:

Использование Kafka в интервью

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

Различия между этими двумя режимами заключаются в том, как потребители взаимодействуют с данными:

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

Вот некоторые ситуации, когда стоит рассмотреть использование очереди сообщений:

• Необходим асинхронный подход к обработке. Например, когда пользователи загружают видео на YouTube, мы можем немедленно сделать видео стандартного разрешения доступным, а затем поместить видео (по ссылке) в Kafka для перекодирования в асинхронном режиме.

• Нужно гарантировать соблюдение порядка обработки сообщений. Подобный подход используется в задаче Ticketmaster, где важно сохранять порядок добавления пользователей в список ожидания.

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

Потоки полезны в следующих случаях:

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

• Несколько потребителей одновременно нуждаются в обработке одних и тех же сообщений. Например, в задаче Live Comments можно использовать Kafka как систему публикации-подписки, направляя комментарии различным потребителям.

Что нужно знать о Kafka для собеседований по проектированию систем

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

Масштабируемость

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

Хотя размер сообщения в Kafka не ограничен жесткими рамками (его можно настроить через параметр message.max.bytes), рекомендуется ограничиваться размером менее 1 МБ для оптимальной производительности, снижения нагрузки на память и оптимизации использования сети.

Хорошее оборудование позволяет одному брокеру хранить примерно 1 ТБ данных и обрабатывать до миллиона сообщений в секунду (хотя реальная цифра зависит от размера сообщений и характеристик оборудования). Если ваша система не требует большего объема, разговор о масштабировании скорее всего неактуален.

Однако, если возникает потребность в масштабировании, есть два основных подхода:

1. Горизонтальное масштабирование с дополнительными брокерами: Простейший способ масштабирования - добавить дополнительные брокеры в кластер. Это помогает распределить нагрузку и повысить устойчивость к отказам. Главное - убедитесь, что ваши темы имеют достаточное число разделов, чтобы воспользоваться новыми брокерами. Больше разделов позволит увеличить параллелизм и улучшить распределение нагрузки. Без достаточного числа разделов новый брокер останется невостребованным.

2. Стратегия разделения данных: Основная задача при масштабировании Kafka - выбрать правильную стратегию распределения данных. Важнейшей частью решения является правильный выбор ключа сообщений. Алгоритм разделения основан на хeшировании ключа (по умолчанию используется хэш-функция MurmurHash2) с последующим вычислением остатка от деления на число разделов (partition = hash(key) % num_partitions). Неправильный выбор ключа может привести к возникновению горячих разделов, перегруженных трафиком. Хорошими ключами являются равномерно распределенные ключи.

При работе с Kafka обычно внимание уделяется масштабированию конкретных тем, а не всего кластера целиком. Это связано с тем, что разные темы могут предъявлять разные требования. Одна тема может требовать большого количества разделов и широкого распределения по брокерам, в то время как другая справляется с нагрузкой на одном брокере.

Обработка горячих разделов

Интервьюеры любят проверять ваше понимание обработки "горячих" разделов. Рассмотрим ситуацию с Ad Click Aggregator, где Kafka собирает потоки событий нажатий пользователями на объявления. Логичным решением было бы разделить события по идентификаторам объявлений. Но предположим, что корпорация X запускает новую рекламную кампанию, и соответствующий раздел оказывается перегруженным ("горячим").

Существует несколько способов справиться с такой ситуацией:

1. Случайное распределение без ключа: Если не указывать ключ, Kafka произвольно назначает раздел сообщению, обеспечивая равномерное распределение. Недостатком является потеря гарантии порядка. Подходит, если порядок не играет роли.

2. Добавление соли (random salting): Ключ сообщения можно дополнить случайным числом или временной меткой. Это улучшит распределение нагрузки, хотя усложняет последующую агрегацию на стороне потребителя.

3. Использование составного ключа: Вместо простого идентификатора объявления используйте комбинацию, включающую географическое положение или сегменты пользователей. Этот подход помогает более равномерно распределять нагрузку и особенно полезен, если вы можете определить атрибуты, которые изменяются независимо от идентификатора объявления.

4. Обратное давление (back pressure): В зависимости от требований, иногда проще замедлить темп поступления сообщений. Managed сервис Kafka может предложить встроенный механизм управления обратным давлением. Если вы используете собственный кластер Kafka, вручную проверяйте задержку раздела (partition lag) и замедляйте работу, если она слишком велика.

Отказоустойчивость и долговечность

Одна из причин выбора Kafka - высокая гарантия сохранности данных. Но как именно Kafka защищает ваши данные и предотвращает потерю сообщений?

Устойчивость данных достигается за счет механизма репликации Kafka. Каждый раздел реплицируется на несколько брокеров, где один выступает в качестве лидера, а остальные - последователей. Сообщение, отправленное производителем, сначала сохраняется на ведущем экземпляре, а затем реплицируется на последователей. Благодаря этому даже при отказе брокера данные сохраняются. Настройки подтверждений производителя играют ключевую роль. Установка режима acks=all гарантирует, что сообщение подтверждается только после того, как все последователи его получили, обеспечивая максимальную надежность.

Настройка коэффициента репликации ваших тем (replication factor) позволяет регулировать уровень отказоустойчивости. Коэффициент репликации равен числу копий каждого раздела. Типичное значение - 3, означающее наличие лидера и двух реплик. Поэтому даже при выходе из строя одного брокера данные останутся доступны на оставшихся, и одна из реплик сможет стать новым лидером.

Что произойдет, если откажет потребитель?

Kafka обычно воспринимается как устойчивая система. Часто говорят: "Kafka всегда доступна, иногда согласована." Вопрос вроде "Что случится, если Kafka выйдет из строя?" не реалистичен, и вы можете мягко возразить интервьюеру, если услышите подобное.

Зато гораздо важнее обсудить ситуацию, когда падает потребитель. В этом случае, механизмы отказоустойчивости Kafka помогают обеспечить непрерывность:

1. Управление смещениями (offsets): Помните, что разделы представляют собой журналы с уникальным смещением для каждого сообщения. Потребители отмечают свои смещения в Kafka после обработки сообщения. Когда потребитель перезапускается, он восстанавливается с последнего зафиксированного смещения, продолжая работу там, где остановился, гарантируя отсутствие потерь и повторной обработки сообщений.

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

Рассматривая компромисс между производительностью и устойчивостью, обратите внимание на выбор подходящего момента для фиксации смещений. Например, в задаче проектирования Web Crawler важно зафиксировать смещение только после успешного сохранения сырого HTML в хранилище BLOB-объектов. Чем сложнее работа потребителя, тем выше вероятность повторной обработки после сбоев. Следовательно, лучше минимизировать сложность действий потребителя и разбить процесс на две фазы: скачивание HTML и последующий парсинг.

Обработка ошибок и повторных попыток (retries)

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

Повторные попытки производителя

Сначала рассмотрим сценарий, когда сообщение не удается отправить в Kafka. Ошибки могут возникать из-за проблем сети, недоступности брокера или временных сбоев. Чтобы смягчить последствия, Kafka-производители поддерживают автоматические повторы. Вот пример конфигурации производителя:

const producer = kafka.producer({
  retry: {
    retries: 5, // Повторить до 5 раз
    initialRetryTime: 100, // Ждать 100мс между повторами
  },
  idempotent: true,
});

Повторные попытки потребителя

Со стороны потребителя ситуация немного сложнее. Поскольку Kafka не поддерживает повторные попытки из коробки(в отличие от AWS SQS) необходимо разработать собственную логику обработки ошибок. Один из популярных подходов - создать отдельную тему для неудачных сообщений и отдельный потребитель для их обработки. Таким образом, можно повторять попытки столько раз, сколько потребуется, не влияя на основной поток. Если какое-то сообщение многократно терпит неудачу, оно перемещается в специальную очередь неисправных сообщений (DLQ - dead letter queue), где его можно проанализировать позднее.

Например, в задаче построения Web Crawler мы выбрали альтернативу Kafka - сервис AWS SQS, позволяющий воспользоваться встроенной функциональностью повторных попыток и DLQ, без необходимости собственной реализации.

Оптимизация производительности

Особенно при использовании Kafka в качестве потока событий важно учитывать производительность, чтобы максимально быстро обрабатывать сообщения.

Первое, что можно сделать - объединять сообщения в пакеты (батчи - batches) перед отправкой в Kafka. Это простая настройка, обеспечивающая увеличение скорости отправки сообщений.

const producer = kafka.producer({
  batch: {
    maxSize: 16384, // Максимальный размер пакета в байтах
    maxTime: 100, // Максимальное время ожидания перед отправкой пакета
  },
});

Еще один эффективный метод ускорения - сжатие сообщений на стороне производителя. Оно реализуется путем установки соответствующего алгоритма сжатия (например, GZIP, Snappy или LZ4). Фактически, мы уменьшаем размер сообщений, ускоряя передачу.

const producer = kafka.producer({
  compression: CompressionTypes.GZIP,
});

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

Политики хранения данных (Retention Policies)

Каждая тема Kafka имеет политику хранения, определяющую срок хранения сообщений в журнале. Параметры настройки включают retention.ms и retention.bytes, и по умолчанию сообщения хранятся 7 дней.

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

Заключение

Поздравляю, вы сделали это! Давайте кратко подведем итоги.

Apache Kafka - это открытая платформа для распределенной потоковой передачи событий, разработанная с учетом требований высокой производительности, масштабируемости и долговечности. Она позволяет производителям отправлять сообщения и читать их потребителям, сохраняя сообщения в упорядоченных, неизменяемых разделах на нескольких брокерах. Kafka идеально подходит для обработки больших объемов данных в реальном времени и асинхронного обмена сообщениями.

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