Задержка Kafka (Kafka lag)

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

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

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

Краткое описание ключевых концепций лага Kafka

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

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

Понимание запаздывания Kafka

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

• Производители (продюсеры) Kafka - это клиентские приложения, которые отправляют сообщения или события.
• Брокеры долго хранят сообщения и делают их доступными в логических партициях, сохраняя порядок.
• Потребители - это клиентские приложения, которые реагируют на сообщения.

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

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

При обработке сообщений потребители Kafka отслеживают последнюю позицию в партиции, из которой они в последний раз читали сообщение. Эта позиция известна как текущее смещение или смещение потребителя в Kafka. Позиция, соответствующая последнему доступному сообщению в партиции, называется log-end offset. Разница между log-end-offset и current-offset называется consumer lag. Отставание Kafka - это разница между последним сообщением, произведенным производителем, и последним сообщением, потребленным группой потребителей.

Причины задержки Kafka

Несколько факторов могут увеличить задержку Kafka.

Внезапный всплеск трафика

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

Проблемы с назначением партиций производителей

Когда приходит новое сообщение, Kafka сначала определяет партицию, в которую оно будет записано. Логика, которую она использует для определения правильного раздела для записи, называется стратегией назначения партиций производителя. Для этого Kafka использует одну из предопределенных стратегий назначения партиций или собственную стратегию. По умолчанию для назначения партиции используется хэш ключа. Существуют и другие стратегии назначения разделов, такие как round-robin, sticky partition и т. д. Kafka также поддерживает пользовательскую стратегию партиций.

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

Назначение разделов для потребителей

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

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

Медленные потребители

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

Мониторинг отставания Kafka

Kafka предоставляет механизм по умолчанию для мониторинга отставания кластера. В файле kafka-consumer-groups.sh содержится подробная информация о задержке для всех партиций. Для просмотра задержки можно использовать следующую команду.

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

В приведенном выше выводе можно увидеть текущее смещение, смещение лог-энда и разницу между ними в виде задержки. Помимо встроенных скриптов, существуют инструменты с открытым исходным кодом для мониторинга отставания Kafka. Burrow - пример инструмента мониторинга Kafka с открытым исходным кодом. Также для мониторинга отставания Kafka можно использовать экспортеры логов, такие как Kafka Lag Exporter или Kafka Exporter, а также общие инструменты мониторинга, такие как Prometheus или Grafana.

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

Уменьшение задержки Kafka

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

Скорость потребления сообщений

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

• fetch.max.bytes: Эта конфигурация контролирует максимальный объем данных, возвращаемых сервером за один запрос. Установка более высоких значений обеспечивает меньшее количество запросов, но более высокую задержку.
• fetch.min.bytes: Эта конфигурация контролирует минимальный объем данных, возвращаемых сервером за один запрос. Установка большего значения приводит к уменьшению количества запросов, но увеличению задержки.
• max.partition.fetch.bytes: Эта настройка контролирует максимальный объем данных, возвращаемых сервером для каждой партиции. Как и в предыдущих конфигурациях, установка больших значений приводит к увеличению задержки.
• fetch.max.wait.ms: Эта конфигурация представляет собой количество времени, которое потребитель ожидает, чтобы заполнить пакет. Установка больших значений приводит к снижению количества запросов, но увеличению задержки.

Стратегия партиций

Стратегия назначения разделов определяет, как потребители Kafka будут сопоставлены с различными партициями. Стратегией назначения партиций по умолчанию является RangeAssignor. Она основана на сортировке потребителей в соответствии с member_id, присвоенным им в начале работы. Присвоенное значение member_id зависит от порядка запуска потребителей. Поэтому эти алгоритмы часто приводят к неоптимальному маппингу партиций потребителей, который не обеспечивает наименьшее отставание.

Вместо этого Kafka предоставляет три другие стратегии назначения разделов, которые можно использовать для оптимизации маппинга разделов. Это RoundRobinAssignor, StickyAssignor и CooperativeStickyAssignor.

• RoundRobinAssignor обеспечивает равное распределение партиций между потребителями, но полагается на порядок сортировки member_id, как и RangeAssignor. Следовательно, он также приводит к неоптимальному маппингу потребителей.
• StickyAssignor сохраняет существующие отображения при создании новых потребителей. Это помогает минимизировать операции перебалансировки.
• CooperativeStickyAssignor аналогичен StickyAssignor, но позволяет потребителям, не затронутым ребалансировкой, обрабатывать сообщения, что играет роль в минимизации задержки.

Параметр partition.assignment.strategy управляет стратегией назначения партиций. Чтобы изменить стратегию назначения партиций на CooperativeStickyAssignor, воспользуйтесь приведенным ниже фрагментом.

Альтернативный вариант - реализовать собственную стратегию назначения партиций, которая учитывает отставание группы потребителей при назначении партиций. Сообщество разработчиков Kafka работает над ее реализацией здесь.

Конфигурации брокера

Производительность брокера Kafka становится узким местом в достижении минимально возможного отставания в системе Kafka. Kafka предоставляет несколько вариантов конфигурации для тонкой настройки брокера. Например, брокер Kafka предоставляет конфигурацию под названием group.initial.rebalance.delay.ms для управления поведением ребалансировки групп. Параметр определяет время, в течение которого брокер ожидает присоединения новых потребителей, прежде чем начать ребалансировку. Установка оптимального значения, которое не оставляет ни одного медленно подключающегося потребителя до начала ребалансировки, помогает сократить количество операций ребалансировки.

Kafka также предоставляет два параметра для управления количеством потоков, используемых при обработке сообщений. Параметр num.network.thread определяет количество сетевых потоков, используемых для обработки запросов от клиентских приложений. При большом количестве одновременных запросов следует увеличить это значение, чтобы потребители не оставались без сообщений из-за перегрузки сети. Параметр num.io.threads управляет количеством потоков для операций ввода и вывода. Значение должно быть как минимум равно количеству процессоров в экземпляре.

Увеличение размера пакета сообщений часто приводит к увеличению задержки потребителей, поскольку потребители получают сразу большое количество сообщений, когда поступают пакетные запросы. Параметр брокера message.max.bytes определяет максимальный размер сообщения, который может поддерживать брокер. Установка этого параметра на меньшее значение, с которым потребители могут легко справиться, помогает уменьшить задержку.

Конфигурации производителей

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

Параметр partitioner.class управляет стратегией партиции для производителей. Можно использовать разбиение RoundRobin или реализовать собственное разбиение, чтобы обеспечить равномерное распределение сообщений по партициям. Kafka также предоставляет механизм, называемый адаптивной партицией, чтобы направлять сообщения в более быстрые партиции. Если для конфигурации partitioner.adaptive.partitioning.enable установлено значение true, Kafka направляет сообщения в те разделы, которые работают быстрее.

Производители, отправляющие большие пакеты сообщений, часто создают задержку на стороне потребителя. Чтобы уменьшить задержку, инженеры должны точно настроить значения параметров batch.size и linger.ms, которые влияют на размер пакетов. Параметр batch.size управляет количеством сообщений в пакете, а linger.ms определяет количество времени, которое Kafka ждет, чтобы заполнить пакет.

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

Лучшие практики для минимизации отставания Kafka

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

Используйте правильный класс разделителя производителей

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

Включите адаптивную партицию

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

Придерживайтесь меньшего размера партии производителя

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

Использование CooperativeStickyAssignor для назначения партиций потребителей

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

Тонкая настройка конфигураций пакетной обработки потребителей

Установка более высоких значений для конфигураций выборки потребителей поможет создать большую партию для потребителей и увеличить пропускную способность. Однако более высокая пропускная способность потребителей не всегда приводит к уменьшению задержки. Если потребители слишком долго ожидают заполнения партий, задержка увеличится. Поэтому при настройке пакетных конфигураций необходимо подобрать оптимальное значение для 'fetch.max.wait.ms'.

Единая стратегия настройки производителей, брокеров и потребителей

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

Оптимизация факторов, связанных с аппаратным обеспечением

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

Заключение

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

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