Как и зачем разворачивать приложение на Apache Spark в Kubernetes02.04.2021 12:03

Теперь запустим демо-приложение, которое считает число π. Для этого мы возьмем репозиторий от Google, где уже есть это приложение:

git clone https://github.com/GoogleCloudPlatform/spark-on-k8s-operator spark-operator

Чтобы запустить приложение, применим YAML-файл:

kubectl apply -f spark-operator/examples/spark-pi.yaml

Проверим, что под создается:

kubectl get pods

Мы видим, что под создается и находится в статусе ContainerCreating:

Теперь посмотрим логи этого пода:

kubectl logs spark-pi-driver | grep 3.1

Мы видим, что приложение отработало и посчитало число π:

Также через Kubernetes мы можем получить список всех Spark-приложений. Это возможно потому, что мы использовали Spark Operator. Kubernetes знает о том, что в нем работает новый тип ресурсов — spark applications.

Итак, получим список всех Spark-приложений:

kubectl get sparkapplications.sparkoperator.k8s.io

В нашем случае это только одно приложение:

Теперь посмотрим различную информацию по этому приложению. Это опять же возможно благодаря Spark Operator:

kubectl describe sparkapplications.sparkoperator.k8s.io spark-pi

Тут мы видим запрошенные ресурсы для Driver и Executor. Также видим образ, который использовался для запуска демо-приложения. И также видим, что мы запускаем приложение типа Scala (еще поддерживаются Python и R):

Ниже видим, какие стадии проходило приложение:

Шаг 4: собираем Docker-образ, запускаем свое приложение

Далее мы будем собирать Docker-образ с собственным приложением. Для этого мы подготовили репозиторий, клонируем его:

git clone https://github.com/stockblog/webinar_spark_k8s/ webinar_spark_k8s

Теперь нам нужно перенести из этого репозитория папкуcustom_jobs в директорию $SPARK_HOME:

mv webinar_spark_k8s/custom_jobs/ $SPARK_HOME

Это нужно, чтобы для сборки образа мы могли использовать утилиту docker-image-tool, которая поставляется вместе со Spark. Она устроена так, что внутрь Docker-context помещает только те файлы и папки, которые находятся внутри $SPARK_HOME. Одно из предназначений этой утилиты — минимизировать размер образа. Но также вы можете собрать собрать образ без нее, просто используя Docker.

Далее мы рассмотрим два простых приложения. Первое читает данные из бакета — выводит 10 первых строк датасета для ознакомления, а второе копирует датасет в другой бакет. Для этого нам понадобятся credentials для доступа к бакету, а также пути к ним. Если у вас еще нет S3-хранилища, вы можете создать его на платформе Mail.ru Cloud Solutions.

Сначала создадим Secret, в нем будут храниться credentials. Вместо PLACE_YOUR_S3_CRED_HERE подставьте ваши данные:

kubectl create secret generic s3-secret 
--from-literal=S3_ACCESS_KEY='PLACE_YOUR_S3_CRED_HERE' 
--from-literal=S3_SECRET_KEY='PLACE_YOUR_S3_CRED_HERE'

Теперь создадим ConfigMap, в котором хранятся пути до бакетов. Вместо S3_PATH и S3_WRITE_PATH подставьте ваши данные:

kubectl create configmap s3path-config 
--from-literal=S3_PATH='s3a://s3-demo/evo_train_new.csv' 
--from-literal=S3_WRITE_PATH='s3a://s3-demo/write/evo_train_csv/'

Теперь познакомимся с содержанием Python-файлов, которые мы будем помещать внутрь приложения. Переходим в папку custom_jobs и открываем файл s3read.py:

cd $SPARK_HOME/custom_jobs
nano s3read.py

Мы видим, что приложение читает данные из бакета и выводит первые 10 строк. Для доступа к бакету мы используем переменные, которые указали ранее: S3_ACCESS_KEY, S3_SECRET_KEY и S3_PATH:

Второе приложение — s3read_write.py. Оно также выводит первые 10 строк из одного бакета и записывает датасет в другой. Для этого используется еще одна переменная S3_WRITE_PATH, в которой хранится путь до второго бакета.

Теперь познакомимся с содержимым docker-файла, который мы будем использовать для сборки:

Он собран на основе официального Docker-файла, который идет в комплекте со Spark. Мы добавили в него JAR-файлы для доступа к S3, без них не получится подключиться к объектному хранилищу. Также мы копируем папку custom_jobs внутрь SPARK_HOME, для того чтобы код приложения попал внутрь Docker-образа.

Теперь будем собирать образ. Мы используем docker-image-tool — инструмент, о котором говорили выше. Мы указали тег webinar_spark_k8s и Docker-файл:

sudo /$SPARK_HOME/bin/docker-image-tool.sh -r $YOUR_DOCKER_REPO -t webinar_spark_k8s -p ~/webinar_spark_k8s/yamls_configs/Dockerfile build

Видим, что образ успешно собрался:

Далее мы опубликуем образ в DockerHub. Для начала зададим переменную DOCKER_REPO, в которой хранится имя репозитория. Вы можете использовать публичный или приватный репозиторий. Вот инструкция, как залогиниться в приватный репозиторий. Мы будем использовать наш публичный репозиторий mcscloud:

export DOCKER_REPO=mcscloud

Теперь отправим образ в репозиторий:

sudo /$SPARK_HOME/bin/docker-image-tool.sh -r $DOCKER_REPO -t webinar_spark_k8s -p ~/webinar_spark_k8s/yamls_configs/Dockerfile push

Теперь, когда мы запушили образ в репозиторий, мы будем использовать его, чтобы развернуть в Spark. Такой подход позволяет нам запускать Spark Jobs стандартным для Kubernetes способом — декларативным. Это означает, что мы декларативно описываем состояние Job, а Kubernetes уже сам будет приводить ее к желаемому состоянию. Это возможно благодаря тому, что мы используем Kubernetes Operator.

Перейдем в директорию yaml_configs и посмотрим на содержимое файла s3read_write_with_secret_cfgmap.yaml. На основе этого файла вы сможете создавать свои:

cd webinar_spark_k8s/yaml_configs
nano s3read_write_with_secret_cfgmap.yaml

В этом файле видно, что мы используем Python третьей версии. Мы используем образ, который только что запушили в репозиторий. Наше приложение s3read_write.py находится внутри этого образа в папке custom_jobs. Далее мы задаем restart policy: если приложение упадет, оно будет перезапущено три раза с интервалом в 10 секунд.

Также мы задаем требования по ресурсам, отдельно для Driver и Executor. Для доступа к S3 мы указываем названия credentials и ConfigMap, в которых у нас хранятся данные для доступа к хранилищу: s3-secret и s3path-config. Они также задаются отдельно для Driver и Executor.

Есть более простой способ указать эти переменные. Рассмотрим файл s3read_example_one.yaml:

Здесь все то же самое, но переменные окружения объявлены явно внутри YAML-файла. Мы не рекомендуем так делать в продуктовых средах, но на этапе тестирования или проверки MVP можно попробовать.

Теперь запустим приложение, применив YAML-файл:

kubectl apply -f ~/webinar_spark_k8s/yamls_configs/s3read_write_with_secret_cfgmap.yaml

Приложение создалось:

Теперь посмотрим список подов в Kubernetes:

kubectl get pods

Видим, что сейчас создается контейнер с приложением:

Теперь снова проверим список Spark-приложений:

kubectl get sparkapplications.sparkoperator.k8s.io

Видим, что приложение только что запустилось, 13 секунд назад:

Опять посмотрим список подов:

kubectl get pods

Видим, что Driver уже работает, сейчас создается Executor:

Если через некоторое время еще раз проверить список подов, мы увидим, что контейнер уже создался и приложение находится в статусе Running, значит, оно работает:

Мы можем выполнить команду kubectl get events и увидеть все события, которые связаны с приложением: как оно стартовало, нет ли ошибок, когда добавлялись поды и так далее. Вы можете посмотреть это сами, мы не будем приводить скриншот.

Через некоторое время снова проверяем поды:

kubectl get pods

Теперь все поды в состоянии Completed, значит, приложение закончило работу:

Посмотрим логи:

kubectl logs s3read-write-test-driver

Видим, что приложение отработало успешно. Оно вывело в лог первые 10 строчек из бакета:

Теперь проверим, что оно записало этот датасет в другой бакет. Напомним, что в ConfigMap мы указали название нашего бакета — s3-demo — и папку, куда нужно складывать файлы, — write/evo_train_csv. Заходим в бакет и убеждаемся, что файлы создались:

Итог: наше кастомное Spark-приложение запустилось и работает в Kubernetes.

Шаг 5: устанавливаем Spark History Server

С нашим приложением есть одна сложность. Если под, на котором оно работает, будет уничтожен, все логи также пропадут. Поэтому мы будем перенаправлять логи в отдельный S3-бакет, а для просмотра будем использовать Spark History Server, который мы также развернем в Kubernetes.

Чтобы разделить приложение и History Server, мы создадим в Kubernetes отдельный Namespace, это как бы виртуальный кластер внутри Kubernetes. Namespace позволяет разделять объекты, ресурсы, доступы и так далее.

Создаем Namespace:

kubectl create ns spark-history-server

Проверим, что он создался командой kubectl get ns:

Для того, чтобы Spark History Server мог читать логи из S3, нам нужно снова создать Secret с нашими данными для доступа к объектному хранилищу. Мы уже создавали такой Secret ранее, но так как History Server находится в другом Namespace, он не сможет получить к нему доступ.

Создаем новый Secret, указываем только что созданный Namespace. Вместо PLACE_YOUR_S3_CRED_HERE укажите свои данные:

kubectl create secret generic s3-secret 
--from-literal=S3_ACCESS_KEY='PLACE_YOUR_S3_CRED_HERE' 
--from-literal=S3_SECRET_KEY='PLACE_YOUR_S3_CRED_HERE' -n spark-history-server

Проверим, что Secret создан с помощью kubectl get secrets -n spark-history-server:

Теперь перейдем непосредственно к установке Spark History Server. Для этого мы снова воспользуемся Helm. Добавим репозиторий:

helm repo add stable https://charts.helm.sh/stable

Чтобы установить Spark History Server, сначала нужно изменить нужные нам настройки. Для этого мы воспользуемся файлом values-hs.yaml. Переходим в папку с файлом и открываем его на редактирование:

cd webinar_spark_k8s/yamls_configs/
nano values-hs.yaml

Мы прокомментируем настройки, чтобы вы знали, что вам нужно будет поправить. Во-первых, активируем S3. Далее указываем бакет и директорию с логами. Далее указываем название Secret, в котором хранятся credentials. Далее указываем тип сервиса — LoadBalancer. Это значит, что History Server будет доступен по внешнему IP:

Еще один важный момент. В этом файле сейчас не указан способ авторизации, поэтому на сервер сможет зайти любой, если узнает его адрес. В продуктиве мы так делать не рекомендуем, но для тестирования можно пока оставить так.

Теперь мы применяем этот файл:

helm install -f ~/webinar_spark_k8s/yamls_configs/values-hs.yaml my-spark-history-server stable/spark-history-server --namespace spark-history-server

Теперь нам нужно узнать, по какому IP-адресу доступен History Server. Для этого выполняем команду:

kubectl get service -n spark-history-server

Видим, что Cluster-IP уже выделен, а внешнего адреса еще нет. Облако генерирует его:

Подождем несколько минут и повторим команду. Теперь мы видим IP-адрес и порт, на который нужно зайти:

Переходим в History Server, в нашем случае по адресу http://87.239.106.171:18080. Мы видим, что он работает, но в нем пока нет логов:

Чтобы они появились, нам нужно немного изменить наше приложение. Мы сделаем так, чтобы оно писало логи в определенный S3-бакет. Для этого мы создали файл s3_hs_server_test.yaml.

Он практически идентичен нашему предыдущему YAML-файлу за тем исключением, что мы добавили секцию sparkConf и в ней указываем, куда нужно писать логи. Опять же, для упрощения мы прописываем наши credentials в явном виде, но в продуктивном использовании так делать не рекомендуется:

Теперь применим этот файл:

kubectl apply -f ~/webinar_spark_k8s/yamls_configs/s3_hs_server_test.yaml

Посмотрим на состояние подов:

kubectl get pods

Видим, что приложение в процессе работы:

Ждем некоторое время, проверяем поды снова. Видим, что запустился и Executor:

Ждем некоторое время и проверяем логи:

kubectl logs s3read-write-hs-server-test-driver

Видим, что приложение отработало и вывело 10 первых строк из бакета:

Переходим в History Server. Видим, что появилась запись о выполненном приложении:

Мы можем нажать на ссылку в столбце App ID и посмотреть все логи:

Тут мы можем получить всю информацию о работе приложения для полноценной отладки.

Результат

Spark в Kubernetes позволяет изолировать рабочие среды, управлять ограничениями приложений и получить гибкую масштабируемую систему. В этой статье мы установили Spark в Kubernetes, обернули свое приложение в Docker-образ и запустили его Native-способом для Kubernetes.

Новым пользователям платформы мы дарим 3000 бонусов после полной верификации аккаунта. Вы сможете повторить сценарий из статьи или попробовать другие наши сервисы.

Что еще почитать по теме:

1. Наш телеграм-канал Вокруг Kubernetes в Mail.ru Group.

2. MLOps без боли в облаке: как развернуть Kubeflow в продашен-кластере Kubernetes.