Как стать автором
Обновить

Spark Structured Streaming Applications on Kubernetes. Опыт ФАСТЕН РУС

Big Data *
Из песочницы
Сегодня я вам расскажу, как нам удалось решить задачу портирования Spark Structured Streaming Applications на Kubernetes (K8s) и реализовать CI стриминга.

Как всё начиналось?


Стриминг выступает ключевым компонентом BI-платформы ФАСТЕН РУС. Real-time данные используются командой дата анализа для построения оперативных отчетов.

Стриминговые приложения реализованы средствами Spark Structured Streaming. Этот фрэймворк обеспечивает удобный API трансформации данных, что отвечает нашим потребностям в плане оперативности доработок.

Сами стримы поднимались на кластере AWS EMR. Таким образом при подъёме нового стрима на кластер выкладывался ssh-скрипт на сабмит Spark-джобы, после чего запускалось приложение. И сначала нас всё вроде бы устраивало. Но с ростом числа стримов становилась всё более очевидной потребность в реализации CI стриминга, что повысило бы автономность команды дата анализа при запуске приложений на доставку данных по новым сущностям.

И сейчас мы рассмотрим, как нам удалось решить эту проблему путём портирования стриминга на Kubernetes.

Почему Kubernetes?


Kubernetes как менеджер ресурсов в наибольшей степени отвечал нашим потребностям. Это и деплой без даунтайма, и широкий набор инструментов реализации CI на Kubernetes, включая Helm. Кроме того, наша команда обладала достаточной экспертизой по реализации CI-пайплайнов на K8s. Поэтому выбор был очевиден.

Как организована модель управления Spark-приложениями на базе Kubernetes?




Клиент запускает spark-submit на K8s. Cоздаётся pod драйвера приложения. Kubernetes Scheduler привязывает pod к ноде кластера. Затем драйвер отправляет запрос на создание pod’s для запуска экзекьютеров, поды создаются и привязываются к нодам кластера. После чего выполняется стандартный набор операций с последующей конвертацией кода приложения в DAG, декомпозицией на stages, разбивкой на таски и их запуском на экзекьютерах.

Эта модель вполне успешно работает при ручном старте Spark-приложений. Однако, подход c запуском spark-submit вне кластера не устраивал нас с точки зрения реализации CI. Требовалось найти решение, которое позволило бы запускать Spark-джобу (выполнять spark-submit) непосредственно на нодах кластера. И здесь нашим требованиям вполне отвечала модель Kubernetes Operator.

Kubernetes Operator как модель управления жизненным циклом Spark-приложения


Kubernetes Operator — концепция управления statefull-приложениями в Kubernetes, предложенная CoreOS, которая предполагает автоматизацию эксплуатационных задач, таких как деплой приложений, рестарт приложений в случае фэйлов, обновление конфигурации приложений. Одним из ключевых паттернов Kubernetes Operator выступает CRD (CustomResourceDefinitions), который предполагает добавление в кластер K8s кастомных ресурсов, что, в свою очередь, позволяет работать с этими ресурсами как с нативными объектами Kubernetes.

Operator представляет собой демон, который живет в pod’е кластера и реагирует на создание / изменения состояния кастомного ресурса.

Рассмотрим эту концепцию применительно к управлению жизненным циклом Spark-приложений.



Пользователь выполняет команду kubectl apply -f spark-application.yaml, где spark-application.yaml — спецификация Spark-приложения. Operator получает объект Spark-приложения и выполняет spark-submit.

Как мы можем видеть, модель Kubernetes Operator предполагает управление жизненным циклом Spark-приложения непосредственно в кластере Kubernetes, что выступало серьезным аргументом в пользу данной модели в контексте решения наших задач.

В качестве Kubernetes Operator для управления стриминговыми приложениями было принято решение использовать spark-on-k8s-operator. Этот оператор предлагает достаточно удобный API, а также отличается гибкостью в настройке политики рестартов Spark-приложений (что достаточно важно в контексте поддержки стриминговых приложений).

Реализация CI


Для реализации CI стриминга использовался GitLab CI/CD. Деплой Spark-приложений на K8s проводился средствами Helm.

Сам пайплайн предполагает 2 стадии:

  • test — проводится проверка синтаксиса, а также рендеринг Helm-шаблонов;
  • deploy — деплой стриминговых приложений на тестовую (dev) и продуктовую (prod) среды.

Рассмотрим подробнее эти стадии.

На стадии test проводится рендеринг Helm-шаблона Spark-приложений (CRD — SparkApplication) значениями, специфичными для сред.

Ключевыми разделами Helm-шаблона выступают:
  1. spark:
    • version — версия Apache Spark
    • image — используемый Docker-образ
  2. nodeSelector — содержит список (key → value), соответствующих лэйблам подов.
  3. tolerations — указывается список толерейшенов Spark-приложения.
  4. mainClass — класс Spark-приложения
  5. applicationFile — локальный путь, где находится jar-файл Spark-приложения
  6. restartPolicy — политика рестартов Spark-приложения
    • Never — завершенное Spark-приложение не рестартуется
    • Always — завершенное Spark-приложение рестартуется независимо от причины остановки
    • OnFailure — Spark-приложение рестартуется только в случае фэйла
  7. maxSubmissionRetries — максимальное число сабмитов Spark-приложения
  8. driver/executor:
    • cores — количество ядер, выделяемых драйверу/экзекьютору
    • instances (используется только для конфигурации экзекьютеров) — число экзекьюторов
    • memory — объём памяти, выделяемой процессу драйвера/экзекьютора
    • memoryOverhead — объём off-heap memory, выделяемой драйверу/экзекьютору
  9. streams:
    • name — наименование стримингового приложения
    • arguments — аргументы стримингового приложения
  10. sink — путь к датасетам Data Lake на S3

После выполнения рендеринга шаблона приложения деплоятся на тестовую среду dev средствами Helm.

Отработал CI-пайплайн.



После чего запускаем джобу deploy-prod — запуск приложений в продакшен.

Убеждаемся в успешном выполнении джобы.



Как мы можем видеть ниже, приложения запущены, поды — в статусе RUNNING.



Заключение


Портирование Spark Structured Streaming Applications на K8s и последующая реализация CI позволили автоматизировать запуск стримов на доставку данных по новым сущностям. Для подъёма очередного стрима достаточно подготовить Merge Request с описанием конфигурации Spark-приложения в yaml-файле значений и при запуске джобы deploy-prod будет инициирована доставка данных в Data Lake (S3). Это решение обеспечило автономность команды дата анализа при выполнении задач, связанных с добавлением в хранилище новых сущностей. Кроме того, портирование стриминга на K8s и, в частности, управление Spark-приложениями средствами Kubernetes Operator spark-on-k8s-operator существенно повысили отказоустойчивость стриминга. Но об этом уже в следующей статье.
Теги:
Хабы:
Всего голосов 7: ↑7 и ↓0 +7
Просмотры 2.1K
Комментарии 3
Комментарии Комментарии 3

Работа

Data Scientist
112 вакансий