Использование YTsaurus Shuffle сервиса
Примечание
Сервис доступен начиная с версии SPYT 2.7.3.
YTsaurus Shuffle сервис нужен для хранения промежуточных данных между стадиями вычислений. В отличие от стандартного Spark Shuffle сервиса, который хранит промежуточные данные в оперативной памяти и во временных директориях на диске, YTsaurus Shuffle сервис хранит их в чанках YTsaurus. Это позволяет повысить устойчивость к падениям стадий распределенных Spark вычислений.
Включение
Прямой запуск задач
При прямом запуске задач в YTsaurus выставьте conf-параметр spark.ytsaurus.shuffle.enabled в true — приложение будет использовать YTsaurus Shuffle сервис вместо нативного Spark shuffle.
Внимание
При включённом dynamic allocation YTsaurus Shuffle сервис обязателен: без spark.ytsaurus.shuffle.enabled=true запуск завершится ошибкой Dynamic allocation requires YTsaurus shuffle service.
Внутренний standalone-кластер
При запуске внутреннего standalone-кластера командой spark-launch-yt доступны два независимых shuffle-сервиса, каждый управляется своей опцией:
| Опция | Shuffle-сервис | По умолчанию |
|---|---|---|
--enable-spark-shuffle / --disable-spark-shuffle |
Нативный внешний Spark shuffle service | включён |
--enable-ytsaurus-shuffle / --disable-ytsaurus-shuffle |
YTsaurus Shuffle сервис | выключен |
Опция --enable-ytsaurus-shuffle лишь делает YTsaurus Shuffle сервис доступным на воркерах. Какой сервис использует приложение, определяется при его запуске параметром spark.ytsaurus.shuffle.enabled (true — YTsaurus Shuffle, иначе — нативный Spark shuffle). Одно приложение должно использовать только один из сервисов.
Внимание
Если приложение запущено с spark.ytsaurus.shuffle.enabled=true, а кластер поднят без опции --enable-ytsaurus-shuffle, на этапе submit ошибки не будет — приложение упадёт в рантайме на первом shuffle-стейдже.
Примечание
Кластер допустимо запустить с обоими выключенными сервисами (--disable-spark-shuffle --disable-ytsaurus-shuffle) — ошибки не будет, но промежуточные данные будут храниться только на экзекьюторах и потеряются при их падении.
Принцип работы
Spark пишет данные в YTsaurus Shuffle сервис под транзакцией. Жизненный цикл этой транзакции контролируется драйвером, который её периодически пингует. Транзакция запускается в момент регистрации нового shuffle, каждый shuffle записывается под отдельной транзакцией. После завершения работы (как успешного, так и неуспешного) с экземпляром shuffle транзакция откатывается, что приводит к удалению всех чанков, которые были под ней записаны. В случае ошибок на драйвере и его аварийного завершения транзакция также откатывается по истечении таймаута. Таймаут конфигурируется параметром spark.ytsaurus.shuffle.transaction.timeout.
Данные, записанные через shuffle сервис, не доступны для чтения через API YTsaurus.
Рекомендации по применению
YTsaurus Shuffle сервис рекомендуется использовать в следующих случаях:
-
Для сложных расчётов, содержащих много промежуточных стадий (Stage). Например, при соединении (join) нескольких крупных таблиц, размер каждой из которых превышает 10M строк. Если помимо сложных расчётов на кластере выполняются также простые вычисления (например, простые агрегации или соединение большой таблицы с маленькой), Shuffle сервис всё равно рекомендуется включать.
-
При использовании режима прямого запуска задач в YTsaurus. Shuffle сервис записывает промежуточные данные в чанки YTsaurus, которые не теряются в случае падения или вытеснения Spark экзекьюторов. Это позволяет не пересчитывать заново этапы, которые были уже посчитаны, но хранились на потерянных экзекьюторах.
Конфигурационные параметры
Все параметры, относящиеся к конфигурации работы с YTsaurus Shuffle сервисом, описаны на странице конфигурационных параметров.