Что на самом деле происходит при запуске задания Spark?

Введение

По мере роста наборов данных эффективно обрабатывать их на одной машине становится сложнее из-за ограничений вычислительной мощности процессора, объёма памяти и пропускной способности подсистемы хранения.

Apache Spark решает эти задачи, предоставляя распределённую систему обработки данных, которая разбивает рабочую нагрузку на разделы и выполняет задачи параллельно на ядрах процессора и нескольких машинах.

Однако Spark — это не просто добавление новых машин.

Чтобы проектировать эффективные и надёжные конвейеры обработки данных, необходимо понимать, как взаимодействуют компоненты Spark: драйверы, исполнители, RDD, DataFrame, разделы, задания, этапы и задачи.

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

1. Зачем нужен Apache Spark: параллельная обработка данных

Apache Spark — это распределённая вычислительная система, предназначенная для крупномасштабной обработки данных и инженерии данных.

Spark обрабатывает большие данные с помощью параллельной обработки:

Рисунок A. Обобщённая архитектура распределённого кластера Apache Spark (кластер, драйвер и рабочие узлы) и сравнение с выполнением на одном узле (создано Kuriko IWAI)

В отличие от обработки на одном узле (справа, рисунок A), Spark разбивает работу на задачи, используя несколько узлов, и выполняет их на доступных ресурсах, в том числе на нескольких машинах, если они доступны.

В данном случае распределённый режим не означает, что использование нескольких машин обязательно. Spark может работать в локальном режиме на одной машине. Даже в этом случае данные можно разделить на разделы и обрабатывать параллельно.

2. Основные компоненты архитектуры Apache Spark

Spark состоит из нескольких основных компонентов:

• Драйвер Spark (драйвер): центральный процесс, координирующий работу приложения Spark.
• Диспетчер кластера: внешняя служба, отвечающая за выделение аппаратных ресурсов, таких как ядра процессора и память.
• Рабочие узлы (рабочий узел): физическая или виртуальная машина либо вычислительный узел в кластере.
• Исполнители: распределённые рабочие процессы, запускаемые на рабочих узлах.
• Java Virtual Machine (JVM): среда выполнения, необходимая для исполнения байт-кода Spark (поскольку Spark написан на Scala/Java).

Сначала драйвер координирует приложение Spark и преобразует его в направленный ациклический граф (DAG) физических задач выполнения, а затем распределяет их между исполнителями.

Затем диспетчер кластера выделяет ресурсы в соответствии с системой управления ресурсами, используемой при развёртывании.

Рабочие узлы предоставляют исполнителям вычислительные ресурсы; исполнители выполняют задачи и хранят данные во время их выполнения.

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

2-1. Пример электронной торговли: структура входных данных

Представим небольшую компанию, занимающуюся электронной торговлей.

У неё есть 10 000 записей о продажах; в каждой указаны артикул (SKU) и количество проданных единиц. Предположим, что возможны три артикула: A, B, и C.

Таблица 1. Пример распределения записей о продажах в электронной торговле и ожидаемых результатов агрегирования.

Наша цель — вычислить общее количество проданных единиц по каждому артикулу:

1. Отфильтровать записи, у которых qty <= 0, затем
2. Сгруппировать данные по артикулам и агрегировать значения.

Предположим, что в этом иллюстративном полном наборе данных после фильтрации останется 9 800 из 10 000 записей. Ожидаемые итоговые значения, которые рассчитает Spark:

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. Кластер Spark

Опишу упрощённый кластер для нашего примера.

Рабочие узлы

Зададим 3 рабочих узла, каждый с 8 vCPU и 32 GiB оперативной памяти:

Рисунок B. Распределение ресурсов между рабочими узлами (создано Kuriko IWAI)

Всего в системе доступно 24 vCPU.

Исполнители Spark и распределение ресурсов ЦП/памяти

Предположим, я выделю:

• 7 исполнителей.
• По 2 ядра vCPU на каждый исполнитель.
• По 3 ГиБ памяти кучи JVM на каждый исполнитель.

Приложение Spark запрашивает у диспетчера кластера ресурсы для этих 7 исполнителей.

Затем диспетчер кластера распределяет ресурсы ЦП и память с учётом использования ресурсов каждым рабочим узлом.

Например, он решает создать 14 слотов для задач — по 2 ядра vCPU на каждый исполнитель × 7 исполнителей:

Рисунок C-1. Ядра vCPU, память и распределение исполнителей между рабочими узлами при неравномерной загрузке рабочего узла A

Обратите внимание: 3 ГиБ памяти, выделенные исполнителю, — это не вся доступная ему память, поскольку необходимо зарезервировать дополнительную память для рабочих узлов и других выполняемых задач.

Предположим, что для одной задачи требуется 1 vCPU (spark.task.cpus = 1), каждый рабочий узел распределяет свои ресурсы vCPU следующим образом:

Рисунок C-2. Параллельное распределение ресурсов vCPU между рабочими узлами

Это показывает два важных момента:

1. Каждый рабочий узел имеет собственный уровень загрузки ЦП (на рисунке C-2 загрузка рабочего узла A составляет 75%, а рабочих узлов B и C — по-прежнему 50%).
2. Хотя в системе есть 24 ядра vCPU (8 vCPU на рабочий узел × 3 рабочих узла), использовать можно только 14 ядер vCPU; остальные 10 остаются незадействованными для этих задач.

2-3. Виртуальная машина Java (JVM) в основе Spark

Apache Spark построен на основе виртуальной машины Java (JVM).

JVM — это среда выполнения, необходимая для исполнения байт-кода Spark.

Поскольку Spark написан на Scala/Java:

• При использовании Scala или Java в приложении: код выполняется непосредственно внутри JVM.
• При использовании PySpark (Python): код выполняется в процессе Python, который обменивается данными с драйвером на базе JVM через мост, например Py4J. Исполнители Spark на рабочих узлах по-прежнему работают внутри JVM, которая затем управляет рабочими процессами Python, чтобы выполнять код Python и пользовательские функции Python (Python UDF).

В обоих случаях JVM управляет потоками выполнения, распределением памяти и сборкой мусора как для драйвера, так и для исполнителей.

JVM драйвера и исполнителей в приложении PySpark

В приложении PySpark драйвер Python взаимодействует с драйвером Spark, работающим в JVM;

• Драйвер Python: среда, в которой пишется и выполняется код PySpark на уровне приложения.
• JVM драйвера: координирует выполнение и планирование задач Spark.
• JVM исполнителя: выполняет задачи Spark и управляет ресурсами выполнения. Если для выполнения задач требуется Python, задействуются рабочие процессы Python.
• JVM клиента (клиентский процесс или компьютер, с которого отправляется задание): JVM, работающая на локальном компьютере или узле периферийного уровня/шлюза.

Как показано на рисунке, JVM клиента сначала подключается к JVM драйвера, которая затем координирует работу с JVM исполнителей:

Рисунок D. Общая схема взаимодействия JVM клиента, драйвера и исполнителей

Далее JVM драйвера размещает драйвер Spark, который управляет приложением, а JVM исполнителей — отдельных исполнителей, отвечающих за выполнение этих задач.

В каждой JVM драйвера есть куча JVM для эффективной обработки данных, кэширования и хранения пользовательских объектов.

2-4. Подробно о стойких распределённых наборах данных (RDD)

Стойкий распределённый набор данных (RDD) — это неизменяемая секционированная коллекция элементов, которую Spark может обрабатывать параллельно.

RDD отслеживают зависимости между преобразованиями, позволяя Spark во многих случаях отказа повторно вычислять утраченные секции.

В названии отражены три свойства:

• Стойкий: Spark записывает происхождение — какие RDD зависят от предыдущих RDD, — и часто может восстановить недостающие секции, не храня копии каждого промежуточного результата.
• Распределённый: набор данных представлен секциями, которые Spark может обрабатывать на разных исполнителях. Наличие нескольких физических машин не обязательно: RDD поддерживаются и в локальном режиме.
• Набор данных: RDD может содержать пары «ключ — значение», целые числа, строки, кортежи и другие поддерживаемые элементы, а не только реляционные строки. API RDD включает такие операции, как map, filter, reduceByKey, join, и aggregateByKey.

RDD — это абстракция логической коллекции и вычислений, а не декларация о том, что определённые разделы навсегда закреплены за определёнными исполнителями.

Разделы

Раздел — это логическая часть данных, которой обычно соответствует одна задача в рамках этапа.

Предположим, мы создаём 100 разделов (выбор именно 100 начальных разделов сделан для наглядности).

Если во входных данных 10 000 элементов, на каждый раздел приходится примерно по 100 элементов, если данные распределены относительно равномерно:

Рисунок E. Параллельное распределение данных по 100 начальным разделам

3. API RDD и API DataFrame: внутренняя оптимизация Catalyst

Исходный пример с RDD помогает понять происхождение данных, разделы и перемешивание данных.

Однако при типичной структурированной обработке ETL лучше начинать с API DataFrame, поскольку он предоставляет столбцы, информацию о схеме, выражения в стиле SQL и операции, доступные оптимизатору.

На схеме ниже показана взаимосвязь:

Рисунок F. Как связаны понятия графов в Spark: происхождение RDD, DataFrame и DAG выполнения Spark

Как у RDD есть граф происхождения, описывающий преобразования, так и план запроса DataFrame представляет его структурированный запрос.

Spark формирует задачи на выполнение на основе зависимостей и физических операторов.

При выполнении DataFrame внутри могут использоваться механизмы на основе RDD, но это не означает, что при написании кода для DataFrame пользователям нужно вручную строить граф происхождения RDD.

3-1. План запроса DataFrame

План запроса DataFrame — это последовательность логических и физических шагов, которую формирует среда выполнения (фиолетовый блок на рисунке L).

Запрос DataFrame проходит через несколько представлений:

1. Логический план: описание запрошенных операций (сканирование, фильтрация, группировка, агрегирование).
2. Анализируемый и оптимизированный логический план: Spark разрешает ссылки и применяет подходящие логические оптимизации, например удаление ненужных столбцов или проталкивание предикатов.
3. Физический план: Spark выбирает исполняемые операторы, например сканирование, частичное и итоговое агрегирование, обмен данными, соединения и сортировку.
4. Выполнение: физический план выполняется с помощью механизмов Spark и в конечном итоге включает задачи на исполнителях.

В нашем примере физический план может включать следующие шаги:

• Частичное хеш-агрегирование,
• обмен данными по SKU и
• итоговое хеш-агрегирование.

3-2. Сравнение действий: API RDD и API DataFrame

Сравним API RDD и API DataFrame на примере агрегирования по SKU.

API RDD

Элементы — это (sku, qty) пары. Логика агрегирования на Python задана явно, а Spark следует зависимостям RDD:

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

API DataFrame

Предположим, что sales_df содержит столбцы sku (строковый тип) и qty (числовой тип):

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

В обоих примерах решается одна и та же бизнес-задача:

Отфильтровать недействительные записи о продажах и агрегировать корректные объёмы по SKU.

И в обоих случаях collect() запускает выполнение Spark и возвращает все полученные записи драйверу.

Ключевое различие заключается в представлении результата:

• API RDD: возвращает объекты Python, например (sku, total_qty) кортежи.
• API DataFrame: возвращает структурированные Row объекты с именованными полями, например sku и total_qty.

Общее сравнение API RDD и API DataFrame:

Таблица 2. Сравнение возможностей и уровня абстракции API RDD и API DataFrame: (создано Kuriko IWAI)

4. Подводя итоги

Цель настройки Spark — не максимизировать количество разделов или загрузку процессора по отдельности.

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

В этой статье мы рассмотрим основные взаимосвязи в системе:

• Ресурсы: Воркеры размещают исполнители; исполнители запускают задачи; заданные ограничения по процессору и памяти определяют степень параллелизма.
• Программирование с помощью RDD: Преобразования создают определения неизменяемых RDD и информацию о происхождении данных; действия запускают вычисления.
• Программирование с помощью DataFrame: Преобразования формируют логические операции запроса; оптимизатор выбирает физические операторы, которые можно выполнить; действия запускают вычисления.
• Выполнение: Spark распределяет работу по заданиям, этапам и задачам, а задачи обрабатывают разделы на исполнителях. Перемешивание данных и другие зависимости создают границы координации между этапами.

Понимание того, как Spark выполняет распределенные рабочие нагрузки, — лишь основа.

Вы можете изучить семиэтапный процесс вычислений с использованием API RDD.