Как выбрать правильное число shuffle-партиций, памяти executor'а и количество ядер?

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

Shuffle-партиции: отталкивайтесь от целевого размера партиции, а не фиксированного числа

spark.sql.shuffle.partitions по умолчанию равен 200 независимо от размера данных, что неверно почти для любой реальной нагрузки — слишком много партиций для маленькой задачи (накладные расходы на планирование задач доминируют) и слишком мало для большой (каждая партиция сбрасывается на диск). Более разумная отправная точка: разделите общий размер входных данных shuffle-стейджа на целевые 128-200 МБ на партицию.

Затем проверьте: после изменения стало ли время выполнения задач внутри стейджа более равномерным (меньше «длиннохвостых» задач) и снизился ли объём сброса на диск. Если сброс всё ещё велик — значения занижены; если большинство партиций теперь почти пусты — завышены.

Память executor'а: рассчитывайте на самую большую ожидаемую партицию, а не на среднюю

OOM у executor'ов почти всегда возникает из-за самой большой партиции в стейдже, а не средней — поэтому расчёт памяти по средней партиции изначально недооценивает потребность. Разумная нижняя граница — размер_самой_большой_партиции * 3-4 (запас на накладные расходы десериализации и любую агрегацию в памяти), ограниченная тем, что реально может предложить тип инстанса вашего кластера на один executor.

Число ядер: больше — не всегда бесплатно

Число ядер executor'а определяет, сколько задач выполняется одновременно в рамках одного бюджета памяти executor'а — если задрать это число слишком высоко, эти задачи начинают конкурировать за один и тот же heap, что проявляется как рост времени GC, а не рост пропускной способности. 4-5 ядер на executor — разумный потолок для большинства JVM-нагрузок; превышение обычно меняет паузы на сборку мусора на параллелизм, который вы фактически не получаете.