Що насправді відбувається під час запуску завдання Spark?

Вступ

Зі збільшенням обсягів наборів даних ефективно обробляти їх на одній машині стає дедалі складніше через обмеження потужності процесора, обсягу пам’яті та пропускної здатності сховища.

Apache 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) та кількість проданих одиниць. Припустімо, що є три можливі SKU: A, B, і C.

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

Наша мета — обчислити загальну кількість проданих одиниць кожного SKU, не враховуючи недійсні записи:

1. Відфільтрувати записи, у яких qty <= 0, а потім
2. Згрупувати записи за SKU та обчислити підсумки.

Для наочності припустімо, що в повному наборі даних після фільтрування залишилося 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 і розподіл ресурсів CPU/пам’яті

Припустімо, я виділю:

• 7 виконавців.
• по 2 ядра vCPU для кожного виконавця.
• по 3 GiB пам’яті купи JVM для кожного виконавця.

Застосунок Spark надсилає менеджеру кластера запит на виділення ресурсів для цих 7 виконавців.

Потім менеджер кластера визначає обсяг CPU та пам’яті для кожного робочого вузла, спираючись на використання ресурсів.

Наприклад, він вирішує створити 14 слотів для завдань — по 2 ядра vCPU на виконавця × 7 виконавців:

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

Зверніть увагу: 3 GiB пам’яті, виділені виконавцю, — це не весь обсяг його пам’яті, оскільки потрібно зарезервувати додаткову пам’ять для робочих вузлів та інших процесів.

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

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

Це демонструє два важливі моменти:

1. Кожен робочий вузол має унікальний рівень використання CPU (на рисунку 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 драйвера та JVM виконавців

У цій схемі JVM драйвера містить драйвер Spark, який керує застосунком, а 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 та DataFrame:

Таблиця 2. Порівняння функцій і рівня абстракції API RDD та API DataFrame (створено Kuriko IWAI)

4. Підсумки

Мета налаштування Spark — не максимізувати кількість розділів або завантаження ЦП окремо одне від одного.

Мета полягає в тому, щоб зменшити загальні витрати й затримку виконання від початку до кінця, не спричиняючи нестабільності.

У цій статті ми дізнаємося про основні взаємозв’язки в системі Spark:

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

Розуміння того, як Spark виконує розподілені робочі навантаження, — лише основа.

Ви можете ознайомитися з семикроковим потоком обчислень за допомогою API RDD.