Кластер Spark

Spark — кластерная вычислительная платформа с открытым исходным кодом для распределенной пакетной и потоковой обработки данных. В рамках сервиса пользователи могут конфигурировать и создавать кластеры Spark для последующего запуска на них соответствующих задач препроцессинга данных. Организация работы с кластером Spark описана в разделах ниже. Пример, иллюстрирующий препроцессинг данных на кластере Spark, размещен на GitHub.

Пользователь создает Jupyter Server конфигурации Spark (CPU) с заданными параметрами. Сервис разворачивает кластер Spark соответствующей конфигурации в Workspace пользователя. Схема создания кластера Spark в заданной конфигурации представлена ниже.

../_images/schm__create-spark-in-given-config.ru.svg

SparkSession и SparkContext

SparkSession, начиная с версии Spark 2.0, является основной точкой входа в программирование Spark с использованием Dataframe и Dataset. В ранних версиях Spark в качестве точки входа в программирование наборов данных RDD и соединения с кластером использовался SparkContext.

Точка входа SparkContext в Spark версий 2.0 и выше имеется для обеспечения обратной совместимости.

Точка входа SparkSession представляет собой комбинированный класс для всех различных контекстов, которые использовались отдельно до версии Spark 2.0, включая SparkContext. Создание экземпляра SparkSession с использованием SparkSession.builder необходимо для начала работы с RDD, DataFrame и Dataset.

Spark Session включает в себя все API, доступные в контекстах:

  • Spark Context

  • SQL Context

Загрузка данных из хранилища S3

Данные, препроцессинг которых нужно осуществить в среде Spark, могут располагаться в хранилище S3. Доступ к хранилищу S3 осуществляется по протоколу S3A. Для получения доступа требуется задать конфигурацию для объекта класса SparkContext в атрибуте _jsc при помощи метода hadoopConfiguration().set(). Адрес endpoint, access key и secret key задаются следующим образом:

sc._jsc.hadoopConfiguration().set("fs.s3.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
sc._jsc.hadoopConfiguration().set("fs.s3a.endpoint", "https://your_endpoint_name")
sc._jsc.hadoopConfiguration().set("fs.s3a.access.key", "your_access_key")
sc._jsc.hadoopConfiguration().set("fs.s3a.secret.key", "your_secret_key")

После указания данных параметров можно обращаться к данным в объектном хранилище S3 по ссылке, включающей в себя имя бакета S3. Например:

path = 's3a://your_bucket_name/path/dataset.csv'

После этого вы можете загрузить данные и создать датасет Spark следующим образом:

df = spark.read.format('csv').load(path)

Чтение данных из хранилища S3 Advanced (OBS)

Подробнее про OBS можно почитать в документации.

Предлагаем использовать следующий скрипт для чтения данных из бакета OBS:

from pyspark.sql import SparkSession
from pyspark.context import SparkContext


if __name__ == "__main__":
    # Create a SparkSession
    print("Start")
#     sc = SparkContext('local')
    sc = SparkContext.getOrCreate()


    spark = SparkSession(sc)
#     sc.setLogLevel("DEBUG");


    folder_path = "s3a://bucket/folder/"
    pattern_date_adv = "filename.parquet"


    config_spark_s3_adv = {
        'access_id': '',
        'access_key': '',
        'impl': 'org.apache.hadoop.fs.s3a.S3AFileSystem',
        'endpoint': 'https://obs.ru-moscow-1.hc.sbercloud.ru'
    }

    spark._jsc.hadoopConfiguration().set("fs.s3a.impl", config_spark_s3_adv["impl"])
    spark._jsc.hadoopConfiguration().set("fs.s3a.endpoint", config_spark_s3_adv["endpoint"])
    spark._jsc.hadoopConfiguration().set("fs.s3a.access.key", config_spark_s3_adv["access_id"])
    spark._jsc.hadoopConfiguration().set("fs.s3a.secret.key", config_spark_s3_adv["access_key"])
#     spark._jsc.hadoopConfiguration().set("fs.s3a.connection.ssl.enabled", "false")
    spark._jsc.hadoopConfiguration().set("fs.s3a.path.style.access", "true") # Check this option!

    dummy_adv_dataframe = spark.read.format('parquet').option("header", "true").load(folder_path + pattern_date_adv)
    print("The dataframe has this many rows:", dummy_adv_dataframe.count())


    # Stop the session
    spark.stop()

Установка дополнительных библиотек при работе с кластером Spark

Подробнее про установку дополнительных библиотек можно узнать в разделе Установка дополнительных библиотек в Jupyter Server.

Запуск препроцессинга данных на кластере Spark

Работа с датафреймами Spark ведется по принципу «ленивых вычислений». Вычисления производятся тогда, когда пользователь запрашивает их результат. Собрать можно с помощью функции collect(), просмотреть состояние датасета — с помощью функции show().

Например, для выполнения SQL-запроса к содержимому датафрейма требуется создать временное представление, затем запустить задание Spark на препроцессинг, и собрать данные:

df.createOrReplaceTempView('temp_view')

result = spark.sql('''
    /* SQL-statement (FROM temp_view) */
''').collect()

Тарификация кластера Spark

Jupyter со Spark с CPU (окружение)
Вычислительные ресурсы
Jupyter со Spark с CPU (окружение) состоит из двух частей
  • Ресурсы Driver — cpu-ai-small (4 vCPU, 16 GB);

  • ресурсы Executor (2 единицы) — cpu-ai-middle (8 vCPU, 32 GB).

Время выполнения операций

11 минут 25 секунд (округляется до целого числа минут в большую сторону)

Расчет стоимости (в рублях с учетом НДС):

\((0,14+0,29 \times 2) \times 12=8,64\)

Где:

  • 0,14 — стоимость CPU-минуты с учетом НДС (по тарифу).

  • 0,24 — стоимость CPU-минуты с учетом НДС (по тарифу).

  • 2 — количество Executor.

  • 12 — количество минут, которые проработала задача.

Масштабирование ресурсов

В данный момент для кластеров Spark не реализовано динамическое масштабирование ресурсов. Это означает, что количество Executor-ов останется неизменным и соответствующим заданному при создании кластера значению на протяжении всего времени существования кластера.

В будущих релизах будет добавлено автоматическое масштабирование кластера — количество Executor-ов будет изменяться до заданного максимального значения в зависимости от нагрузки на кластер.

Запуск задачи обучения, используя Spark

Используйте образ cr.msk.sbercloud.ru/aicloud-base-images/spark-3.2.1:0.0.32.1 для запуска задачи обучения Spark. Подробнее Job.