Привет, Хабр! Некоторое время назад я выкладывала статью об улучшении пропускной способности коннекторов Kafka Connect. Сегодня хочу представить вам перевод технического документа от Confluent об оптимизации развертывания самого Kafka кластера, оригинал документа можно скачать по ссылке.
Этот документ посвящен тому, как определить, что именно нужно оптимизировать (пропускная способность, задержка, надежность или доступность), как настроить развертывание Kafka в соответствии с этими целями и убедиться в их достижении с помощью мониторинга.
Сегодня я выкладываю первую часть, в которой обсуждается, как определить цели оптимизации и улучшить пропускную способность Kafka. Последующие статьи будут посвящены задержке, надежности, доступности, мониторингу и бенчмаркингу.
Введение
Не внося никаких изменений в параметры конфигурации Kafka, вы можете создать среду Kafka для разработки и протестировать базовую функциональность. Однако тот факт, что Kafka работает прямо из коробки, не означает, что вы не захотите провести некоторую настройку перед запуском в производство.
Чтобы оптимизировать работу в соответствии с требованиями сервиса, необходимо изменить некоторые параметры конфигурации Kafka.
Чтобы убедиться, что развертывание Kafka оптимизировано для ваших целей, вам обязательно нужно настроить некоторые параметры конфигурации и провести сравнительный анализ в вашей собственной среде. В идеале это нужно сделать до запуска в производство или, по крайней мере, до масштабирования на кластер большего размера.
Как решить, что следует оптимизировать
Первый шаг - решить, какие цели обслуживания вы хотите оптимизировать. Мы рассмотрим четыре цели, которые часто связаны с компромиссами между собой: пропускная способность, задержка, надежность и доступность.
Чтобы понять, какие цели вы хотите оптимизировать, подумайте, для каких вариантов использования предназначен ваш кластер. Подумайте о приложениях, бизнес-требованиях - тех вещах, которые гарантировано не должны дать сбой в этих сценариях использования. Подумайте, как Kafka в качестве технологии потоковой передачи событий вписывается в конвейер вашего бизнеса.
Иногда на вопрос о том, какую цель сервиса следует оптимизировать, трудно ответить, но вы должны заставить свою команду обсудить исходные сценарии использования в бизнесе и их основные цели. Это обсуждение важно по двум причинам.
Первая причина заключается в том, что вы не можете достичь всех целей одновременно. Иногда приходится искать компромисс между пропускной способностью, задержкой, надежностью и доступностью, о чем мы подробно рассказываем в этом техническом документе. Возможно, вы знакомы с широко известным компромиссом между пропускной способностью и задержкой, а также, возможно, между надежностью и доступностью (Прим. переводчика - cм. статью о расширенной CAP теореме, напрмер, здесь). Рассматривая всю систему в целом, вы поймете, что не можете думать ни об одной из них в отдельности, поэтому в данной серии статей рассматриваются все четыре цели обслуживания вместе. Это не значит, что оптимизация одной из этих целей приводит к полной потере остальных. Это означает, что все они взаимосвязаны, а значит, вы не можете максимизировать их все одновременно.
Вторая причина, по которой важно определить, какую цель сервиса вы хотите оптимизировать, заключается в том, что вы можете и должны настроить параметры конфигурации Kafka для ее достижения. Вам необходимо понять, чего именно ждут от системы ваши пользователи, чтобы оптимизировать Kafka для удовлетворения их потребностей.
Хотите ли вы добиться высокой пропускной способности, т. е. скорости передачи данных от продюсеров к брокерам или от брокеров к консьюмерам? В некоторых случаях скорость составляет миллионы записей в секунду. Благодаря дизайну Kafka, записывать в нее большие объемы данных совсем несложно. Это быстрее, чем пытаться прогнать объемы данных через традиционную базу данных или хранилища типа ключ-значение, и это можно сделать с помощью скромного оборудования.
Хотите ли вы добиться низкой задержки, которая представляет собой время, затрачиваемое на передачу сообщений из одной точки в другую (от продюсеров к брокерам и консьюмерам)? Одним из примеров использования, где требуется низкая задержка, является приложение для чата, где получатель сообщения должен получить его с минимальной задержкой. Другие примеры - интерактивные веб-сайты, на которых пользователи следят за сообщениями друзей по сети, или обработка потоков в реальном времени для Интернета вещей.
Хотите ли вы оптимизировать работу для обеспечения высокой надежности, которая гарантирует, что сообщения, которые были зафиксированы, не будут потеряны? Одним из примеров использования, где необходима высокая надежность, является конвейер микросервисов с потоковой передачей событий, использующий Kafka в качестве хранилища событий. Другой вариант - интеграция между источником потоковой передачи событий и постоянным хранилищем (например, AWS S3) для критически важного бизнес-контента.
Хотите ли вы оптимизировать систему для обеспечения высокой доступности, которая минимизирует время простоя в случае непредвиденных сбоев? Kafka - это распределенная система, и она разработана таким образом, чтобы быть устойчивой к сбоям. В случаях использования, требующих высокой доступности, важно настроить Kafka таким образом, чтобы она восстанавливалась после сбоев как можно быстрее.
Если аспекты, которые вы хотите оптимизировать, относятся ко всем топикам в кластере Kafka, вы можете установить параметры конфигурации для всех брокеров, чтобы они применялись ко всем топикам. С другой стороны, если вы хотите оптимизировать разные аспекты в разных топиках, вы можете задать переопределения для топиков по некоторым параметрам конфигурации. Без явного переопределения параметров топика применяется значение конфигурации брокера.
Существуют сотни различных параметров конфигурации, и в этом техническом документе представлено очень небольшое их количество, имеющее значение для обсуждения. Названия, описания и значения параметров по умолчанию актуальны для Confluent Platform версии 5.5 (Kafka версии 2.5). Дополнительные сведения об этих параметрах конфигурации, переопределении параметров топика и других параметрах конфигурации см. в документации.
Прежде чем мы перейдем к рассмотрению того, как оптимизировать Kafka для различных целей обслуживания, сделаем одно предостережение: Значения некоторых параметров конфигурации, обсуждаемых в этой статье, зависят от других факторов, таких как средний размер сообщения, количество партиций и т. д. Они могут сильно различаться в зависимости от окружения. Для некоторых параметров конфигурации мы приводим диапазон разумных значений, но помните, что для проверки настроек для конкретного развертывания всегда важно использовать бенчмаркинг.
Оптимизация пропускной способности
Чтобы оптимизировать пропускную способность, продюсерам, брокерам и консьюмерам необходимо переместить как можно больше данных за определенное время. Для достижения высокой пропускной способности вы пытаетесь максимизировать скорость перемещения этих данных. Скорость передачи данных должна быть как можно выше. Топик - это единица параллелизма в Kafka. Сообщения в разные партиции могут параллельно отправляться продюсерами, параллельно записываться разными брокерами и параллельно читаться разными консьюмерами. Как правило, большее количество партиций топиков приводит к увеличению пропускной способности, а для максимизации пропускной способности необходимо достаточное количество партиций, чтобы задействовать все брокеры в кластере.
Хотя может показаться заманчивым просто создавать топики с большим количеством партиций, увеличение количества партиций имеет свои недостатки. Просмотрите рекомендации Confluent по выбору количества партиций. Тщательно выбирайте количество партиций, ориентируясь на пропускную способность продюсера и консьюмера, а также на эталонную производительность в вашей среде. Также обратите внимание на структуру шаблонов данных и назначение ключей, чтобы сообщения распределялись по партициям топика как можно более равномерно. Это позволит предотвратить перегрузку некоторых партиций относительно других.
Далее обсудим стратегию пакетной обработки продюсеров Kafka. Продюсеры могут группировать сообщения, направляемые в одну и ту же партицию, то есть собирать несколько сообщений для отправки в одном запросе. Самый важный шаг, который вы можете предпринять для оптимизации пропускной способности, - это настроить пакетную обработку продюсера, чтобы увеличить размер пакета и время, затрачиваемое на ожидание заполнения пакета сообщениями. Больший размер партии приводит к уменьшению количества запросов к брокерам, что снижает нагрузку на продюсеров, а также нагрузку на CPU брокера для обработки каждого запроса. В Java-клиенте вы можете настроить параметр batch.size для увеличения максимального размера в байтах каждого пакета сообщений. Чтобы дать партиям больше времени на заполнение, вы можете настроить параметр linger.ms, чтобы продюсер ждал дольше перед отправкой. Задержка позволяет продюсеру подождать, пока партия достигнет настроенного размера batch.size. Компромисс заключается в более высокой задержке, поскольку сообщения отправляются не сразу, как только они готовы к отправке.
Вы также можете включить сжатие, что означает, что большое количество битов может быть отправлено в виде меньшего количества битов. Включите сжатие, настроив параметр compression.type, который может принимать значение одного из следующих стандартных кодеков сжатия: lz4, snappy, zstd и gzip. Для повышения производительности мы обычно рекомендуем lz4 и избегаем gzip, так как он может перегружать CPU. Сжатие применяется к полным пакетам данных, поэтому эффективность пакетирования также влияет на степень сжатия - большее наполнение пакетов приводит к лучшему коэффициенту сжатия. Когда брокер получает сжатый пакет сообщений от продюсера, он всегда распаковывает данные, чтобы проверить их. После этого он учитывает кодек сжатия топика назначения.
Если кодек сжатия в топике назначения оставлен по умолчанию (таким же как у продюсера) или если кодеки пакета и топика назначения одинаковы, брокер получает сжатый пакет от клиента и записывает его непосредственно в файл журнала топика, не выполняя циклов повторного сжатия данных.
В противном случае брокеру придется повторно сжимать данные в соответствии с кодеком топика назначения, что может привести к снижению производительности, поэтому по возможности оставьте кодеки сжатия одинаковыми.
Декомпрессия и рекомпрессия также могут происходить, если продюсеры используют версию до 0.10, поскольку оффсеты должны быть перезаписаны, или если требуется любое другое преобразование формата сообщения. Но в остальном, если вы используете обновленные клиенты и версия log.message.format.version актуальна, преобразование сообщений не требуется.
Когда продюсер отправляет сообщение в кластер Kafka, оно отправляется брокеру-лидеру в целевую партицию. Затем продюсер ожидает ответа от брокера-лидера, чтобы узнать, что его сообщение было зафиксировано, и только после этого приступает к отправке следующих сообщений. В систему встроена автоматическая проверка, гарантирующая, что консьюмеры не могут читать сообщения, которые еще не были зафиксированы. Время отправки брокерами-лидерами ответов может повлиять на пропускную способность продюсера: Чем быстрее продюсер получает ответ, тем быстрее он может отправить следующее сообщение, что обычно приводит к увеличению пропускной способности. Поэтому продюсеры могут установить конфигурационный параметр acks, чтобы указать количество подтверждений, которые должен получить брокер-лидер, прежде чем ответить продюсеру подтверждением. Установка acks=1 заставляет брокер-лидер производить запись в свой локальный журнал и затем подтверждать запрос, не дожидаясь подтверждения от всех брокеров-последователей. Компромисс заключается в том, что вам придется мириться с меньшей надежностью, поскольку продюсеру не нужно ждать, пока сообщение будет реплицировано другим брокерам.
Продюсеры Kafka автоматически выделяют память для Java-клиента для хранения неотправленных сообщений. Если этот лимит памяти достигнут, то продюсер будет блокировать дополнительные отправки, пока не освободится память или пока не пройдет время max.block.ms. Вы можете настроить объем выделяемой памяти с помощью параметра конфигурации buffer.memory. Если у вас не так много партиций, возможно, вам вообще не придется настраивать этот параметр. Однако если у вас много партиций, вы можете настроить параметр buffer.memory (принимая во внимание размер сообщения, время ожидания и количество партиций), чтобы поддерживать конвейерную обработку на большем количестве партиций, что, в свою очередь, позволит лучше использовать пропускную способность большего количества брокеров.
Аналогичным образом вы можете настроить консьюмеров на более высокую пропускную способность, отрегулировав количество данных, получаемых ими при каждой выборке из брокера-лидера. Вы можете увеличить количество данных, получаемых консьюмерами от лидера для каждого запроса выборки, увеличив параметр конфигурации fetch.min.bytes. Этот параметр задает минимальное количество байт, ожидаемое от консьюмера при получении ответа. Увеличение этого показателя также уменьшит количество запросов к брокеру, что снизит нагрузку на процессор брокера при обработке каждого запроса, а значит, повысит пропускную способность. Как и в случае с увеличением размера пакетов для продюсера, при увеличении этого параметра для консьюмера возможен компромисс в виде более высокой задержки. Это происходит потому, что брокер не будет отправлять консьюмеру новые сообщения до тех пор, пока в запросе на выборку не будет достаточно сообщений для удовлетворения размера запроса на выборку, т. е. fetch.min.bytes, или до истечения времени ожидания, т. е. параметра конфигурации fetch.max.wait.ms.
Если приложение позволяет, используйте группы консьюмеров с несколькими консьюмерами для распараллеливания потребления. Распараллеливание потребления может повысить пропускную способность, поскольку несколько консьюмеров могут сбалансировать нагрузку, обрабатывая несколько партиций одновременно. Верхним пределом такого распараллеливания является количество партиций в топике.
Наконец, если у вас есть Java-клиент, вам следует учесть еще два момента. Если вы используете SSL/TLS, поддерживаются как Java 8, так и Java 11, но в идеале следует использовать Java 11, чтобы воспользоваться улучшениями производительности SSL, которые были представлены в Java 9.
Кроме того, следует настроить JVM так, чтобы минимизировать время паузы для сборки мусора (GC). GC необходим, потому что он удаляет неиспользуемые объекты и освобождает память. Однако длительные паузы в GC нежелательны, поскольку они могут негативно повлиять на пропускную способность и в худшем случае привести к "мягкому" сбою брокера, например, из-за таймаута сессии ZooKeeper. Для получения дополнительной информации об оптимальной настройке JVM GC ознакомьтесь с рекомендациями по развертыванию от Confluent.
Сводка конфигураций для оптимизации пропускной способности
Продюсер:
batch.size: увеличить до 100000 – 200000 (по умолчанию 16384)linger.ms: увеличить до 10 - 100 (по умолчанию 0)compression.type=lz4(по умолчанию none, т.е. без сжатия)acks=1(по умолчанию 1)buffer.memory: увеличить, если партиций много (по умолчанию 33554432).
Консьюмер:
fetch.min.bytes: увеличить до ~100000 (по умолчанию 1).