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

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

Обзор архитектуры

Разделы Kafka используются для упорядочивания записей. Производители производят записи, и потребители потребляют их. Производители отправляют записи брокерам Kafka, которые затем хранят данные. Каждый рабочий узел в кластере HDInsight — это брокер Kafka.

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

Репликация используется для дублирования секций между узлами. Разделение защищает от отказов узлов (брокеров). Одна секция между группой реплик обозначается в качестве лидера секции. Трафик производителя направляется к лидеру каждого узла, используя состояние, которым управляет ZooKeeper.

Определение сценария

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

• Высокая пропускная способность, низкая задержка. Для этого сценария требуется высокая пропускная способность и низкая задержка (~100 миллисекунд). Примером этого типа приложения является мониторинг доступности служб.
• Высокая пропускная способность, высокая задержка. Для этого сценария требуется высокая пропускная способность (~1,5 ГБИТ/с), но может допускать более высокую задержку (< 250 мс). Примером этого типа приложения является прием данных телеметрии для практически в реальном времени процессов, таких как приложения обнаружения безопасности и вторжений.
• Низкая пропускная способность, низкая задержка. В этом сценарии требуется низкая задержка (< 10 мс) для обработки в режиме реального времени, но может снизить пропускную способность. Примером этого типа приложения является проверка орфографии и грамматики в Сети.

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

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

Размер партии

Производители Apache Kafka собирают группы сообщений (называемые пакетами), которые отправляются в виде единицы для хранения в одной секции хранилища. Размер пакета означает количество байтов, которые должны присутствовать перед передачей этой группы. batch.size Увеличение параметра может повысить пропускную способность, что снижает затраты на обработку запросов сети и операций ввода-вывода. При легкой нагрузке увеличение размера пакета может увеличить задержку отправки Kafka, так как производитель ожидает, чтобы пакет был готов. При тяжелой нагрузке рекомендуется увеличить размер пакета для улучшения пропускной способности и снижения задержки.

Продюсер требует подтверждения

Требуемая конфигурация производителя acks определяет количество подтверждений, необходимых лидеру секции перед завершением запроса на запись. Этот параметр влияет на надежность данных и принимает значения 0, 1или -1. Значение -1 означает, что подтверждение должно быть получено от всех реплик перед завершением записи. Параметр acks = -1 обеспечивает более надежные гарантии потери данных, но также приводит к увеличению задержки и снижению пропускной способности. Если вашим требованиям к приложению необходима более высокая пропускная способность, попробуйте установить acks = 0 или acks = 1. Имейте в виду, что неучёт всех реплик может снизить надёжность данных.

Сжатие

Производитель Kafka можно настроить для сжатия сообщений перед отправкой их брокерам. Параметр compression.type задает используемый кодек сжатия. Поддерживаемые кодеки сжатия: gzip, snappy и lz4. Сжатие полезно и следует учитывать, если есть ограничение емкости диска.

Среди двух часто используемых кодеков сжатия, gzip и snappy, кодек gzip имеет более высокий коэффициент сжатия, что приводит к снижению использования дисков за счет более высокой загрузки ЦП. Кодек snappy обеспечивает меньше сжатия с меньшими затратами на ЦП. Вы можете решить, какой кодек следует использовать на основе ограничений на диск брокера или ЦП производителя. gzip может сжимать данные в скорости пять раз выше snappy.

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

Параметры брокера

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

Число дисков

Диски хранилища имеют ограниченные IOPS (операции ввода-вывода в секунду) и чтение и запись байтов в секунду. При создании новых разделов Kafka сохраняет каждую новую секцию на диске с наименьшими существующими секциями, чтобы сбалансировать их по доступным дискам. Несмотря на стратегию хранения, при обработке сотен реплик секций на каждом диске Kafka может легко насыщать доступную пропускную способность диска. Компромисс между пропускной способностью и стоимостью. Если вашему приложению требуется более высокая пропускная способность, создайте кластер с большим количеством управляемых дисков на каждый брокер. HDInsight в настоящее время не поддерживает добавление управляемых дисков в работающий кластер. Дополнительные сведения о настройке количества управляемых дисков см. в статье "Настройка хранилища и масштабируемости для Apache Kafka в HDInsight". Узнайте о затратах на увеличение объема дискового пространства для узлов в кластере.

Количество тем и разделов

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

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

Увеличение плотности разделов (число разделов на брокер) приводит к дополнительным затратам, связанным с операциями метаданных и запросами/ответами между лидером раздела и его ведомыми. Даже в отсутствие потока данных реплики секций по-прежнему извлекают данные из лидеров, что приводит к дополнительной обработке для отправки и получения запросов по сети.

Для кластеров Apache Kafka 2.1 и 2.4, как ранее отмечалось в HDInsight, рекомендуется не более 2000 разделов на брокер, включая реплики. Увеличение числа партиций на брокер уменьшает пропускную способность и может также привести к недоступности темы. Дополнительные сведения о поддержке секций Kafka см. в официальной записи блога Apache Kafka о увеличении числа поддерживаемых секций в версии 1.1.0. Дополнительные сведения об изменении тем см. в разделе Apache Kafka: изменение тем.

Количество реплик

Более высокий коэффициент репликации приводит к дополнительным запросам между лидером секции и последователями. Следовательно, более высокий коэффициент репликации потребляет больше дисков и ЦП для обработки дополнительных запросов, увеличения задержки записи и уменьшения пропускной способности.

Рекомендуем использовать для Kafka в Azure HDInsight трехкратное резервирование. Большинство регионов Azure имеют три домена сбоя, но в регионах с двумя доменами сбоя пользователям следует использовать четырёхкратную репликацию.

Дополнительные сведения о репликации см. в разделе Apache Kafka: репликация и Apache Kafka: увеличение коэффициента репликации.

Конфигурации потребителей

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

Количество потребителей

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

Если число потребителей больше числа секций, то вы тратите ресурсы потребителей впустую, поскольку эти потребители простаивают.

Избегайте частого перебаланса потребителей

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

Потребители считаются живыми, если они могут отправить пульс брокеру в пределах session.timeout.ms. В противном случае потребитель считается мертвым или не работает. Эта задержка приводит к перераспределению потребителей. Чем ниже потребитель session.timeout.ms, тем быстрее мы можем обнаружить эти сбои.

session.timeout.ms Если это слишком низко, потребитель может столкнуться с повторяющихся ненужных перебалансов, из-за таких сценариев, как когда пакет сообщений занимает больше времени для обработки или когда приостановка сборки JVM занимает слишком много времени. Если у вас есть потребитель, который тратит слишком много времени на обработку сообщений, вы можете устранить это, увеличив верхний предел на время, которое потребитель может быть неактивным, прежде чем получать больше записей с max.poll.interval.ms помощью или уменьшая максимальный размер пакетов, возвращаемых с параметром max.poll.recordsконфигурации.

Пакетирование

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

Настройка ОС Linux с высокой нагрузкой

Карты памяти

vm.max_map_count определяет максимальное количество mmap для процесса. По умолчанию на виртуальной машине linux кластера Apache Kafka для HDInsight значение равно 65535.

В Apache Kafka каждый сегмент журнала требует пары файлов index/timeindex, и каждый из этих файлов использует одну mmap. Другими словами, каждый сегмент журнала использует два mmap. Таким образом, если каждая секция размещает один сегмент журнала, требуется не менее двух mmap. Количество сегментов журнала на раздел зависит от размера сегмента, интенсивности нагрузки, политики хранения, периода обновления и, как правило, составляет более одного. Mmap value = 2*((partition size)/(segment size))*(partitions)

Если необходимое vm.max_map_countзначение mmap превышает значение, брокер вызовет исключение Map failed .

Чтобы избежать этого исключения, выполните приведённые ниже команды, чтобы проверить размер mmap на ВМ и увеличить его при необходимости на каждом рабочем узле.

# command to find number of index files:
find . -name '*index' | wc -l

# command to view vm.max_map_count for a process:
cat /proc/[kafka-pid]/maps | wc -l

# command to set the limit of vm.max_map_count:
sysctl -w vm.max_map_count=<new_mmap_value>

# This will make sure value remains, even after vm is rebooted:
echo 'vm.max_map_count=<new_mmap_value>' >> /etc/sysctl.conf
sysctl -p

Дальнейшие действия

• Обработка триллионов событий в день с помощью Apache Kafka в Azure
• Что такое Apache Kafka в HDInsight?