С выходом Apache Spark 3.1 в марте 2021-го проект Spark on Kubernetes официально перешел в статус общедоступного и готового к эксплуатации. Это стало результатом трехлетней работы быстрорастущего сообщества, участники которого помогали в разработке и внедрении (изначально поддержка Spark on Kubernetes появилась в Spark 2.3 в феврале 2018 года). Команда Kubernetes aaS от Mail.ru Cloud Solutions перевела самое важное из статьи об основных возможностях Spark 3.1, в которой автор подробно остановился на улучшениях в Spark on Kubernetes.
Полезные источники информации:
- Введение в использование Kubernetes в качестве менеджера ресурсов для Spark (вместо YARN): Pros & Cons of Running Spark on Kubernetes.
- Техническое руководство по успешной работе со Spark on Kubernetes: Setting Up, Managing & Monitoring Spark on Kubernetes.
- Про Data Mechanics: Spark on Kubernetes Made Easy: How Data Mechanics Improves On The Open-Source Version.
Путь Spark on Kubernetes: от бета-поддержки в 2.3 до нового стандарта в 3.1
С выходом Spark 2.3 в начале 2018 года Kubernetes стал новым диспетчером для Spark (помимо YARN, Mesos и автономного режима) в крупных компаниях, возглавляющих проект: RedHat, Palantir, Google, Bloomberg и Lyft. Сначала поддержка имела статус экспериментальной, функций было мало, стабильность и производительность были невысокими.
С тех пор сообщество проекта получило поддержку многочисленных крупных и маленьких компаний, которых заинтересовали преимущества Kubernetes:
- Нативная контейнеризация. Упаковывайте зависимости (и сам Spark) с помощью Docker.
- Эффективное совместное использование ресурсов и ускорение запуска приложений.
- Обширная Open Source-экосистема уменьшает зависимость от облачных провайдеров и вендоров.
В проект было внесено несколько нововведений: от базовых требований вроде поддержки PySpark и R, клиентского режима и монтирования томов в версии 2.4 до мощных оптимизаций вроде динамического выделения (в 3.0) и улучшения обработки выключения нод (в 3.1). За последние три года вышло больше 500 патчей, сильно повысивших надежность и производительность Spark on Kubernetes.
График внесения улучшений в Spark с 2018 по 2021 годы
В результате в новых Spark-проектах в 2021 году Kubernetes все чаще рассматривается в роли стандартного менеджера ресурсов: это следует из популярности Open Source-проекта оператора Spark on Kubernetes и объявлений крупных вендоров, внедряющих Kubernetes вместо Hadoop YARN.
С выходом Spark 3.1 проект Spark on Kubernetes получил статус общедоступного и готового к эксплуатации. В этом релизе было внесено больше 70 исправлений и улучшений производительности. Давайте рассмотрим самые важные функции, которых с нетерпением ждали заказчики.
Улучшенная обработка выключения нод: постепенное отключение исполнителя (новая функция в Spark 3.1)
Эту функцию (SPARK-20624) реализовал Holden Karau, и пока что она доступна только для автономных развертываний и в Kubernetes. Называется функция улучшенная обработка выключения нод, хотя еще одно подходящее название постепенное отключение исполнителя (Graceful Executor Decommissioning).
Эта функция повышает надежность и производительность Spark при использовании Spot-нод (вытесняемые ноды в GCP). Перед остановкой спота с него перемещаются shuffle-данные и содержимое кэша, поэтому влияние на работу Spark-приложения оказывается минимальное. Раньше, когда система убивала спот, все shuffle-файлы терялись, поэтому их приходилось вычислять заново (снова выполнять потенциально очень долгие задачи). Новая фича не требует настройки внешнего shuffle-сервиса, для которого нужно по запросу запускать дорогие ноды хранения и который совместим с Kubernetes.
Новая функция Spark предотвращает внезапное уничтожение спотов
и постепенно выключает исполнитель без потери драгоценных
данных!
Что делает эта функция?
- Исполнитель, который нужно выключить, вносится в черный список: драйвер Spark не будет назначать ему новые задачи. Те задачи, которые сейчас исполняются, не будут принудительно прерваны, но если они сбоят из-за остановки исполнителя, то перезапустятся в другом исполнителе, как и сейчас, а их сбой не будет учитываться в максимальном количестве сбоев (новинка).
- Shuffle-файлы и кэшированные данные мигрируют из выключаемого исполнителя в другой. Если другого нет, например, мы выключаем единственный исполнитель, то можно настроить объектное хранилище (вроде S3) в качестве запасного.
- После завершения миграции исполнитель умирает, а Spark-приложение продолжает работать как ни в чем не бывало.
Когда работает эта функция?
- При использовании Spot / вытесняемых нод облачный провайдер уведомляет о выключении за 60120 секунд. Spark может использовать это время для сохранения важных shuffle-файлов. Этот механизм используется и в тех случаях, когда экземпляр на стороне провайдера по каким-то причинам выключают, допустим, при обслуживании EC2.
- Когда нода Kubernetes пустеет, например для технического обслуживания, или когда вытесняется под исполнителя Spark, например более высокоприоритетным подом.
- Когда убранный исполнитель является частью динамического выделения при уменьшении размера системы из-за простоя исполнителя. В этом случае тоже будут сохранены кэш и shuffle-файлы.
Как включить функцию?
-
С помощью конфигурационных флагов. Нужно включить четыре основных флага Spark:
- spark.decommission.enabled;
- spark.storage.decommission.rddBlocks.enabled;
- spark.storage.decommission.shuffleBlocks.enabled;
- spark.storage.decommission.enabled.
Другие доступные настройки рекомендую поискать прямо в исходном коде.
-
Для получения предупреждения от облачного провайдера о скором отключении ноды, например из-за выключения спота, нужна специфическая интеграция.
Новые опции с томами в Spark on Kubernetes
Начиная со Spark 2.4 при использовании Spark on Kubernetes можно монтировать три типа томов:
- emptyDir: изначально пустая директория, существующая, пока работает под. Полезна для временного хранения. Можно поддерживать ее с помощью диска ноды, SSD или сетевого хранилища.
- hostpath: в ваш под монтируется директория прямо из текущей ноды.
- Заранее статически создаваемый PersistentVolumeClaim. Это Kubernetes-абстракция для разных типов персистентного хранилища. Пользователь должен создать PersistentVolumeClaim заранее, существование тома не привязано к поду.
В Spark 3.1 появилось два новых варианта: NFS и динамически создаваемый PersistentVolumeClaims.
NFS это том, который могут одновременно использовать несколько подов и который можно заранее наполнить данными. Это один из способов обмена информацией, кодом и конфигурациями между Spark-приложениями либо между драйвером и исполнителем внутри какого-нибудь Spark-приложения. В Kubernetes нет NFS-сервера, вы можете запустить его самостоятельно или использовать облачный сервис.
После создания NFS вы можете легко монтировать этот том в Spark-приложение с помощью таких настроек:
spark.kubernetes.driver.volumes.nfs.myshare.mount.path=/sharedspark.kubernetes.driver.volumes.nfs.myshare.mount.readOnly=falsespark.kubernetes.driver.volumes.nfs.myshare.options.server=nfs.example.comspark.kubernetes.driver.volumes.nfs.myshare.options.path=/storage/shared
NFS (Network File System) популярный способ обмена данными
между любыми Spark-приложениями. Теперь он работает и поверх
Kubernetes
Второй новый вариант динамический PVC. Это более удобный способ использования персистентных томов. Раньше нужно было сначала создавать PVC, а затем монтировать их. Но при использовании динамического выделения вы не знаете, сколько можно создать исполнителей в ходе работы вашего приложения, поэтому старый способ был неудобен. К тому же приходилось самостоятельно вычищать ненужные PersistentVolumeClaims либо смиряться с потерей места в хранилище.
Со Spark 3.1 все стало динамическим и автоматизированным. Когда вы инициализируете Spark-приложение или в ходе динамического выделения запрашиваете новые исполнители, в Kubernetes динамически создается PersistentVolumeClaims, который автоматически предоставляет новый PersistentVolumes запрошенного вами класса хранилища. При удалении пода ассоциированные с ним ресурсы автоматически удаляются.
Другие функции Spark 3.1: PySpark UX, стейджинговая диспетчеризация, повышение производительности
В Spark 3.1 появилось два крупных улучшения в UX для разработчиков на PySpark:
- Документация PySpark полностью переделана, теперь она больше соответствует Python и удобна для использования.
- Теперь поддерживаются подсказки типов: получение в IDE бесплатного автозавершения кода и обнаружения статических ошибок.
Автозавершение кода в PySpark с выходом Apache Spark
3.1
Spark History Server, который отображает интерфейс Spark после завершения вашего приложения, теперь может показывать статистику выполненных вами запросов на Structured Streaming.
Стейджинговая диспетчеризация (SPARK-27495) применима только для YARN- и Kubernetes-развертываний при включенном динамическом выделении. Она позволяет вам управлять в коде количеством и типом запрашиваемых для исполнителя ресурсов с точностью на уровне стадий. В частности, вы можете настроить приложение под использование исполнителей с процессорными ресурсами в ходе первой стадии (например, при выполнении ETL или подготовке данных), а в ходе второй стадии на использование видеокарт (скажем, для моделей машинного обучения).
В Spark 3.1 улучшили производительность shuffle хеш-соединения, добавили новые правила для прерывания подвыражений и в оптимизатор Catalyst. Для пользователей PySpark будет полезно то, что в Spark теперь применяется колоночный формат in-memory хранения Apache Arrow 2.0.0 вместо 1.0.2. Это повысит скорость работы приложений, особенно если вам нужно преобразовывать данные между Spark и фреймами данных Pandas. Причем все эти улучшения производительности не потребуют от вас менять код или конфигурацию.
Что еще почитать по теме:
