Spark: shuffle, джойны, перекос

Apache Spark это движок распределённой обработки данных: один и тот же код на DataFrame или SQL выполняется на кластере из сотен машин. На собесе почти всегда спрашивают три вещи: ленивые вычисления, shuffle и что делать с перекосом данных.

Архитектура

• Driver: процесс, где выполняется ваша программа: строит план, делит его на задачи, раздаёт их и собирает результаты.
• Executors: процессы на узлах кластера, выполняют задачи и хранят кешированные данные.
• Cluster manager: выдаёт ресурсы (YARN, Kubernetes, standalone).

Данные разбиты на партиции; одна задача (task) обрабатывает одну партицию на одном ядре. Число партиций определяет параллелизм.

RDD, DataFrame, Dataset

• RDD: низкоуровневая распределённая коллекция объектов. Spark не знает структуры данных и не может оптимизировать.
• DataFrame: таблица с колонками и схемой. Запросы проходят через оптимизатор Catalyst (перестановка фильтров, проталкивание предикатов в источник, отбор колонок, выбор стратегии джойна) и исполняются эффективно (Tungsten, генерация кода).
• Dataset: типизированный DataFrame в Scala и Java.

В PySpark используют DataFrame и SQL. Код на RDD и Python UDF обходит оптимизатор и медленнее.

Ленивые вычисления

Преобразования (select, filter, join, groupBy) только строят план. Вычисление запускают действия (actions): count, collect, write, show. Каждое действие исполняет план заново, если промежуточный результат не закеширован.

План делится на стадии (stages) по границам shuffle; стадия состоит из задач по партициям. Посмотреть план: df.explain(), а фактическое выполнение в Spark UI.

Узкие и широкие преобразования

• Узкие (narrow): каждая выходная партиция зависит от одной входной: filter, select, withColumn. Выполняются без обмена данными.
• Широкие (wide): данные с одинаковым ключом нужно собрать вместе: groupBy, join, distinct, orderBy, repartition. Требуют shuffle.

Shuffle: каждая задача пишет свои данные по ключам в файлы, следующая стадия читает нужные куски со всех узлов по сети. Это диск, сеть и сериализация: самая дорогая операция Spark. Оптимизация обычно сводится к тому, чтобы делать меньше shuffle и перегонять через него меньше данных: фильтровать и отбирать колонки до джойна, агрегировать раньше.

Число партиций после shuffle задаёт spark.sql.shuffle.partitions (по умолчанию 200). Для маленьких данных это слишком много мелких задач, для огромных слишком мало. AQE (adaptive query execution, включён по умолчанию с Spark 3.2) по статистике выполнения объединяет мелкие партиции, меняет стратегию джойна и разбивает перекошенные партиции.

Стратегии джойна

• Broadcast hash join: маленькая таблица целиком копируется на каждый executor, большая не перемешивается. Самый быстрый вариант, если маленькая сторона действительно маленькая. Порог автоматического выбора: spark.sql.autoBroadcastJoinThreshold (по умолчанию 10 МБ); можно подсказать хинтом broadcast(df).
• Sort-merge join: обе стороны перемешиваются по ключу и сортируются. Стандарт для двух больших таблиц.
• Shuffle hash join: перемешивание и хеш-таблица по меньшей стороне в каждой партиции.

Перекос данных (skew)

Если один ключ встречается в миллионы раз чаще других, после shuffle одна партиция огромная. 199 задач закончились за минуту, одна работает час или падает по памяти. Признак в Spark UI: медиана времени задач мала, максимум огромен.

Что делать:

• AQE skew join: Spark сам режет перекошенные партиции при sort-merge join.
• Broadcast, если одна сторона небольшая: shuffle по ключу не нужен вовсе.
• Соление ключа (salting): к горячему ключу в большой таблице добавляют случайный суффикс 0..N-1, а маленькую сторону размножают на N копий с каждым суффиксом. Нагрузка горячего ключа делится на N задач.
• Обработать горячие ключи отдельно (например, NULL-ключи, которые часто и создают перекос).
• Для агрегаций: двухэтапная агрегация, сначала по солёному ключу, потом по исходному.

repartition, coalesce, кеш

• repartition(n) или repartition(col): полный shuffle, равномерные партиции или партиции по ключу.
• coalesce(n): уменьшает число партиций без shuffle, склеивая соседние. Хорош перед записью, чтобы не плодить мелкие файлы, но coalesce(1) сводит всю работу стадии к одной задаче.
• cache() и persist(): сохранить результат, если он используется несколькими действиями. Без повторного использования кеш только тратит память. Не забывайте unpersist().

Частые ошибки

• collect() на большом DataFrame: все данные едут на driver, он падает по памяти.
• Python UDF там, где есть встроенные функции: данные сериализуются в Python и обратно, оптимизатор слепнет.
• Джойн двух больших таблиц без фильтрации и отбора колонок.
• groupByKey на RDD вместо reduceByKey: все значения ключа перегоняются через shuffle без частичной агрегации.
• 200 партиций по умолчанию на терабайтах или на мегабайтах.

Что спросят на собесе

• Driver и executor, партиции, задачи, стадии.
• Преобразования и действия, ленивость.
• Узкие и широкие преобразования, что такое shuffle и почему он дорог.
• Broadcast join и sort-merge join, порог broadcast.
• Как найти и победить перекос: AQE, broadcast, соление.
• repartition против coalesce, когда кешировать.