Масштабирование Spark: как правильно настроить executors и шардирование

Запускаете Spark-задачу, а она либо виснет на стадии Shuffle, либо «съедает» всю память кластера, вылетая с ошибкой OutOfMemory? Знакомо. Многие думают, что для ускорения работы достаточно просто добавить больше узлов в кластер или выделить каждому исполнителю максимум RAM. На деле же избыточные ресурсы часто приводят к тормозам из-за бесконечной сборки мусора в JVM или огромных накладных расходов на управление тысячами мелких задач.

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

Золотое правило настройки ядер: почему 5 - это магическое число?

Казалось бы, чем больше ядер (cores) мы отдадим одному исполнителю, тем быстрее он обработает данные. Но здесь в игру вступает Garbage Collection (GC). Когда один огромный процесс JVM управляет десятками ядер, паузы на очистку памяти становятся слишком длинными, и приложение буквально замирает.

Оптимальным выбором для большинства задач считается установка spark.executors.cores в значение 5. Почему именно столько? Это «золотая середина»: с одной стороны, вы получаете хороший параллелизм внутри одного процесса, с другой - не перегружаете JVM настолько, чтобы сборщик мусора стал главным тормозом системы. Если поставить меньше 5 ядер, вы рискуете увеличить количество операций ввода-вывода. Если больше - количество исполнителей в кластере упадет, и общая мощность системы снизится.

Исключение - совсем маленькие задачи на пару гигабайт. Тут можно ограничиться и одним ядром. Но если в коде есть тяжелые join или groupBy, возвращайтесь к стратегии «5 ядер на executor».

Считаем память для Executors без ошибок

Память в Spark - это не просто одна цифра. Нужно учитывать, что часть ресурсов заберет операционная система, менеджер ресурсов (например, YARN) и так называемый memory overhead (внеблочная память).

Давайте разберем реальный пример. Допустим, у вас на узле 64 ГБ оперативной памяти и вы планируете запустить 3 исполнителя на этот узел. Кажется, что можно дать каждому по 21 ГБ (64 / 3). Но YARN обычно забирает около 10% на свои нужды. В итоге полезной памяти останется примерно 19 ГБ. Если вы попытаетесь выделить все 21 ГБ, менеджер ресурсов просто «убьет» ваш процесс за превышение лимита.

Важный нюанс: не пытайтесь выделить 20 ГБ памяти для executor-а с одним ядром. Это крайне неэффективно. Ресурсы должны расти пропорционально: больше ядер - больше памяти. Хорошим стартом для среднего проекта будет связка: 5 ядер и 20 ГБ RAM на один исполнитель.

Динамическая vs Статическая аллокация: что выбрать?

Если ваш пайплайн запускается по расписанию раз в сутки и обрабатывает предсказуемый объем данных, используйте статическую аллокацию. Вы один раз прописываете spark.executor.instances = 6, spark.executor.cores = 5 и spark.executor.memory = 20g. Это гарантирует стабильность в продакшене: вы точно знаете, сколько ресурсов зарезервировано.

Для интерактивной работы (например, в Jupyter Notebook), где вы то запускаете тяжелый запрос, то просто исследуете данные, спасением станет Dynamic Allocation. Эта функция позволяет Spark самому запрашивать дополнительные исполнители, когда очередь задач растет, и отдавать их обратно в кластер, когда они простаивают.

Чтобы включить этот режим, установите spark.dynamicAllocation.enabled = true. Но будьте внимательны: для этого должен быть запущен внешний сервис shuffle (например, spark.shuffle.service.enabled = true), иначе при удалении простаивающего исполнителя вы потеряете все его промежуточные данные, и Spark придется пересчитывать их заново.

Шардирование и борьба с «перекосом» данных

Правильный тюнинг ресурсов бесполезен, если данные распределены плохо. В Spark это называется Partitioning. Если одна партиция окажется намного больше остальных (data skew), один исполнитель будет работать часами, пока остальные 99 будут просто ждать. Это классическая проблема «длинного хвоста».

Оптимальный размер одной партиции - от 100 до 200 МБ. Почему не 1 КБ? Потому что создание миллионов мелких задач создаст колоссальную нагрузку на Driver. Почему не 10 ГБ? Потому что такая партиция просто не влезет в память executor-а, и Spark начнет сбрасывать данные на диск (spill to disk), что замедлит работу в десятки раз.

Особое внимание уделите параметру spark.sql.shuffle.partitions. По умолчанию он равен 200. Для небольших данных этого много, но для серьезных ETL-задач с огромными джойнами этого катастрофически мало. Попробуйте увеличить его до 400 или даже 1000, чтобы разбить данные на более мелкие, управляемые куски при перетасовке.

Оптимизация Shuffle: как меньше гонять данные по сети

Shuffle - самая дорогая операция в Spark. Это момент, когда данные физически перелетают с одного сервера на другой. Чтобы снизить нагрузку на сеть, используйте map-side агрегацию. Вместо того чтобы отправлять все сырые данные для группировки, Spark сначала частично суммирует или агрегирует их прямо внутри партиции, и только потом пересылает результат.

Если вы видите, что сборка мусора (GC) занимает слишком много времени, попробуйте увеличить spark.executor.memoryOverhead. Это память вне основной кучи JVM (off-heap), которая используется для внутренних нужд Spark. Начните с 1-3 ГБ. Это поможет избежать частых падений процесса при интенсивном обмене данными.

Еще один профессиональный совет: смените стандартный Java-сериализатор на Kryo. Он работает значительно быстрее и компактнее, что напрямую снижает объем данных, которые нужно передавать по сети во время Shuffle.

Итоговый чек-лист для настройки ETL-пайплайна

Если вы настраиваете регулярный процесс обработки данных, ориентируйтесь на этот шаблон конфигурации:

• Количество исполнителей: исходя из объема данных (начните с 6-10).
• Ядра на исполнитель: 5 (spark.executor.cores = 5).
• Память на исполнитель: 20 ГБ (spark.executor.memory = 20g).
• Дополнительная память: 3 ГБ (spark.executor.memoryOverhead = 3g).
• Память драйвера: 8 ГБ (spark.driver.memory = 8g).
• Партиции при shuffle: от 400 (spark.sql.shuffle.partitions = 400).

Следующие шаги по оптимизации

Когда вы настроите ресурсы и партиции, следующим этапом станет оптимизация самого кода. Обратите внимание на использование broadcast joins для маленьких таблиц - это позволит полностью избежать Shuffle-операций. Также изучите адаптивный запросный планировщик (AQE - Adaptive Query Execution), который в новых версиях Spark умеет автоматически менять количество партиций прямо во время выполнения задачи, основываясь на реальном размере данных.