Понимание параметров Apache Spark - пошаговое руководство для настройки и оптимизации Apache Spark
Всего несколько параметров позволять Вам существенно увеличить производительность Spark Streaming приложений, которые читают топики с Kafka и записывают их в хранилище(например в HDFS).
Первым, что стоит оптимизировать, это получателя сообщений с Kafka. Таким получателем выступает Ваше Spark Streaming приложение, которое читает сообщения с топика Kafka порциями(блоками, партициями). Размер этих блоков нужно настраивать относительно предположительной скорости сохранения данных в хранилище. По умолчанию размер этих блоков не установлен и его нужно определить исходя из скорости обработки сообщений Вашим получателем.
Вот формула по которой можно понять, сколько же сообщений за раз получит Ваш воркер:
После нескольких запусков Вы уже приблизительно понимаете сколько записей в секунду обрабатывает Ваш получатель. К примеру мы знаем, что воркер справляется приблизительно с 1000 записей в секунду. Возьмём половину от этого и установим значение 500 для параметра spark.streaming.receiver.maxRate.
Далее можно приступать к настройке параметра spark.streaming.blockInterval. Этот параметр контролирует размер партиции. Если поставить параметр в 200ms, то это будет означать, что каждые 200ms воркер будет получать блок сообщений. 200ms означает что в 1 секунду будет создано 5 блоков. С значением maxRate=500 это будет означать что 1 блок будет содержать 100 сообщений с топика(очереди сообщений).
Первым, что стоит оптимизировать, это получателя сообщений с Kafka. Таким получателем выступает Ваше Spark Streaming приложение, которое читает сообщения с топика Kafka порциями(блоками, партициями). Размер этих блоков нужно настраивать относительно предположительной скорости сохранения данных в хранилище. По умолчанию размер этих блоков не установлен и его нужно определить исходя из скорости обработки сообщений Вашим получателем.
Вот формула по которой можно понять, сколько же сообщений за раз получит Ваш воркер:
partitionSize = (1000 / blockInterval) * maxRate
После нескольких запусков Вы уже приблизительно понимаете сколько записей в секунду обрабатывает Ваш получатель. К примеру мы знаем, что воркер справляется приблизительно с 1000 записей в секунду. Возьмём половину от этого и установим значение 500 для параметра spark.streaming.receiver.maxRate.
Далее можно приступать к настройке параметра spark.streaming.blockInterval. Этот параметр контролирует размер партиции. Если поставить параметр в 200ms, то это будет означать, что каждые 200ms воркер будет получать блок сообщений. 200ms означает что в 1 секунду будет создано 5 блоков. С значением maxRate=500 это будет означать что 1 блок будет содержать 100 сообщений с топика(очереди сообщений).
После нескольких проверок Вы будете понимать какой размер партиции будет подходить для Вас. Нужно помнить, что если Вы увеличиваете значение параметра maxRate, то Вам нужно уменьшать значение параметра blockInterval.
Колличество ядер
При запуске Spark приложения вы можете указать колличество ядер и колличество запущеных екземпляров. Если Ваш Spark работает на Yarn-e, то Вы можете отдельно установить эти параметры (--num-executors 3 --executor-cores 2 если вы хотите использовать 6 ядер для воркера). Если Вы используете Mesos или режим Spark standalone, то указать колличество ядер можно через параметр: --total-executor-cores 6.
Настройка Batch interval
Batch interval - это промежуток времени после истечения которого будет создан новый batch с данными и указывается в секундах. Обычно советуют ставить 5-10 секунд. Выходит, что данные к воркеру попадают не сразу, а акумулируются и отправляются пачкой раз в несколько секунд в виде RDD.
Этот интервал советуют ставить исходя из колличества партиций на которые делиться batch данных. Советуют ставить колличество партиций в 2-3 раза больше чем доступно ядер. Например, если у Вас в наличии 20 ядер, то кол. партиций будет где-то 40-60. Возьмём среднее значение 50. Это значение будет основанием с которого Вы можете начать настраивать кол. создаваемых партиций в каждом интервале.
Теперь, зная кол. партиций и blockInterval можно вычислить batch interval: blockInterval * number partitions. В нашем случае это будет 200ms * 50 = 10000ms(10 сек.). Нужно поексперементировать с этим значением в сторону увеличение и уменьшение так как это приблизительное значение. Большой интервал может объяснятся достаточно длинной задержкой при сохранении данных в Ваше хранилище(например Hive или HBase etc.)
Проверка изменений
Теперь Вы изменили достачно настроек для начала тестирования. Используйте Spark UI для мониторинга запущеных воркеров(Обычно находиться на 4040 порту по адресу http://localhost:4040). Первым делом нужно смотреть на длительность выполнение операций(Stages). Она должна быть меньше чем установленный Вами batch interval.
Также полезно посмотреть на вкладку "Streaming" где должна быть аналитика по времени выполнения операций. Приблизительно время выполнения в 50% случаев должно занимать 85% от установленного batch interval и в 75% случаев оно должно занимать 95% времени от установленного batch interval.
Проверяйте регулярно вкладку "Streaming". Общая задержка не должна быть больше пары минут и среднее время выполнения операций должно быть меньше установленного batch interval. Напомню, что 85% выделенного времени до следующего batch-а с данными будет подходящим значением.
Полезные ссылки:
Коментарі
Дописати коментар