Spark JOIN тормозит из-за одной горячей партиции

В Spark UI 199 задач стадии JOIN завершились за минуту, одна работает уже сорок. Увеличение числа executors почти не меняет время: одна задача не распараллелится только потому, что рядом появились свободные ядра. Частая причина — перекос ключа JOIN. Например, у половины заказов customer_id=0, потому что источнику неизвестен клиент. Все строки с этим ключом попадают в один участок shuffle.

Сначала проверьте, действительно ли дело в skew. В Spark UI сравните длительность задач, объём shuffle read, spill на диск и размеры входных разделов. В данных посчитайте частоты ключей JOIN. Если один ключ встречается в миллионах строк, а медианный ключ — десятки раз, распределение уже подсказывает причину длинного хвоста. Но большой shuffle сам по себе не доказывает skew: медленный executor или внешний источник тоже может затянуть отдельную задачу.

Проверьте смысл «горячего» ключа

Значение 0 может означать «клиент неизвестен». Если обе стороны JOIN имеют много таких строк, соединение по 0 создаст произведение количества строк: тысяча слева на тысячу справа даст миллион совпадений. Это не только проблема производительности, но и неправильный бизнес-результат. Определите, можно ли исключить неизвестный ID из соединения или требуется иное правило сопоставления. Нельзя лечить план запроса до проверки смысла результата.

Если правая таблица действительно мала, Spark может разослать её как broadcast и избежать большого shuffle на этой стороне. Сравните план и объём фактических данных: таблица после фильтрации может стать маленькой, даже если исходная большая. Broadcast не помогает, если результат JOIN на горячем ключе сам гигантский. Если обе стороны велики, сначала проверьте Adaptive Query Execution.

Что делает AQE и чего не делает

В Spark 4.1 оптимизация skewed sort-merge JOIN в AQE может разделять перекошенную shuffle-партицию и, при необходимости, реплицировать соответствующую сторону. Она включается при активных spark.sql.adaptive.enabled и spark.sql.adaptive.skewJoin.enabled, но классифицирует перекос по размеру партиции и порогам, а не по самому факту долгой задачи. Посмотрите итоговый физический план и метрики, прежде чем считать, что механизм сработал.

При устойчивом горячем ключе и больших таблицах иногда используют salting: добавляют к этому ключу несколько искусственных корзин и размножают подходящие строки маленькой стороны по корзинам. Так нагрузка распределяется, но растёт объём данных и сложность запроса. Нельзя случайно добавить разный salt в две стороны и потерять совпадения. Этот приём требует проверки количества строк и контрольной суммы результата на небольшом наборе.

Увеличение spark.sql.shuffle.partitions часто помогает мелким равномерным задачам, но не делит один конкретный ключ, если его хеш по-прежнему отправляет все строки в одну партицию. Правильная последовательность: установить горячий ключ, проверить кардинальность соединения, выбрать стратегию исполнения и сравнить план с результатом. Производительность считается улучшением только при сохранении корректного числа строк.

Например, таблица заказов содержит 10 миллионов строк, из них 4 миллиона с customer_id=0. Справочник клиентов содержит одну строку с этим ID. Результат соединения по нулю даст 4 миллиона строк — это может быть формально корректным техническим JOIN, но не означает, что заказы принадлежат одному клиенту. Если справочник содержит несколько дублей нулевого ID, результат раздуется ещё сильнее. До настройки AQE проверьте уникальность ключа справочника и бизнес-правило для неизвестного клиента.

После изменения стратегии измерьте не только wall time, но и spill, объём shuffle и разницу между самой быстрой и самой медленной задачами. Если длинный хвост исчез, а общий объём shuffle вырос в несколько раз, решение может оказаться дорогим на других датах. Тестируйте на периоде с обычной нагрузкой и на дне, когда горячий ключ максимален. Автоматический план Spark может меняться вместе со статистикой данных.

Для salting отдельно проверьте корректность outer JOIN. Размножение строк на одной стороне и выбор корзины на другой меняют число потенциальных совпадений; при unmatched-строках можно неожиданно получить несколько копий. Сделайте маленький контрольный набор, где известны пары, отсутствующие ключи и ожидаемое число строк. Такой тест показывает и ошибку в логике, и реальную цену распределения горячего ключа.

В курсе Data Engineering Fast‑Track вы работаете со Spark, партиционированием и проверкой конвейера. Посмотрите программу, чтобы разбирать медленный JOIN от плана до распределения ключей.

Отдельно стоит разобрать раздувание суммы при JOIN: там корень ошибки в зерне таблиц, и ускорение Spark её не исправит.