Co tak naprawdę dzieje się podczas uruchamiania zadania Spark?

Wprowadzenie

Wraz ze wzrostem zbiorów danych ich wydajne przetwarzanie na jednym komputerze może stać się trudne ze względu na ograniczenia mocy obliczeniowej procesora, pamięci i przepustowości pamięci masowej.

Apache Spark pomaga sprostać tym wyzwaniom, oferując rozproszony silnik przetwarzania danych, który dzieli zadania na partycje i wykonuje je równolegle na rdzeniach procesora oraz wielu maszynach.

Spark nie polega jednak po prostu na dodawaniu kolejnych maszyn.

Zrozumienie współdziałania jego komponentów — sterowników, executorów, RDD, DataFrame’ów, partycji, zadań, etapów i jednostkowych operacji — jest niezbędne do projektowania wydajnych i niezawodnych potoków przetwarzania danych.

W tym artykule omówię architekturę Sparka, mechanizmy wykonywania zadań i najważniejsze kwestie związane z wydajnością, posługując się praktycznym przykładem agregowania danych o sprzedaży w handlu elektronicznym.

1. Dlaczego Apache Spark — równoległe przetwarzanie danych

Apache Spark to silnik przetwarzania rozproszonego przeznaczony do przetwarzania danych na dużą skalę i inżynierii danych.

Spark przetwarza duże zbiory danych z wykorzystaniem przetwarzania równoległego:

Rysunek A. Architektura klastra rozproszonego Apache Spark z lotu ptaka (klaster, sterownik i węzły robocze) oraz porównanie z wykonywaniem na pojedynczym węźle (utworzył Kuriko IWAI)

W przeciwieństwie do przetwarzania na pojedynczym węźle (po prawej, rysunek A), Spark dzieli pracę na zadania wykonywane na wielu węzłach i rozdziela je między dostępne zasoby różnych maszyn.

W tym przypadku rozproszenie nie oznacza, że wiele maszyn jest niezbędnych. Spark może działać lokalnie na jednym komputerze. Nawet wtedy dane można podzielić na partycje i przetwarzać równolegle.

2. Podstawowa architektura Apache Spark

Spark składa się z kilku kluczowych komponentów:

• Sterownik Spark (sterownik): centralny proces koordynujący działanie aplikacji Spark.
• Menedżer klastra: zewnętrzna usługa odpowiedzialna za przydzielanie zasobów sprzętowych, takich jak rdzenie procesora i pamięć.
• Węzły robocze (węzeł roboczy): fizyczna lub wirtualna maszyna albo węzeł obliczeniowy w klastrze.
• Executory: rozproszone procesy robocze uruchamiane na węzłach roboczych.
• Java Virtual Machine (JVM): środowisko wykonawcze niezbędne do uruchamiania kodu bajtowego Sparka (ponieważ Spark napisano w Scali i Javie).

Najpierw sterownik koordynuje działanie aplikacji Spark i przekształca ją w skierowany graf acykliczny (DAG) zadań wykonywanych fizycznie, a następnie planuje ich wykonanie na executorach.

Następnie menedżer klastra przydziela zasoby zgodnie z systemem zarządzania zasobami używanym w danym wdrożeniu.

Węzły robocze udostępniają executorom zasoby obliczeniowe; executory wykonują zadania i przechowują dane w trakcie ich wykonywania.

Niektóre zarządzane środowiska, takie jak Databricks, używają jednego executora na węzeł roboczy, ale nie jest to uniwersalna zasada Sparka. W zależności od konfiguracji klastra na jednym węźle roboczym może działać kilka executorów.

2-1. Przykład z handlu elektronicznego — struktura danych wejściowych

Wyobraźmy sobie niewielką firmę handlującą przez internet.

Ma ona 10 000 rekordów sprzedaży, z których każdy zawiera jednostkę magazynową (SKU) i sprzedaną ilość. Załóżmy, że możliwe są trzy kody SKU: A, B, oraz C.

Tabela 1. Przykład zestawienia rekordów sprzedaży w handlu elektronicznym i oczekiwanych agregacji.

Naszym celem jest obliczenie łącznej liczby prawidłowo sprzedanych sztuk dla każdego SKU:

1. Odfiltruj rekordy, których qty <= 0, a następnie
2. Zagreguj próbki według SKU.

Załóżmy, że w przykładowym pełnym zbiorze danych po filtrowaniu pozostaje 9800 z 10 000 rekordów. Oczekiwane końcowe sumy obliczone przez Spark wynoszą:

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

2-2. Klaster Spark

Zdefiniuję uproszczony klaster na potrzeby naszego przykładu.

Węzły robocze

Ustawię 3 węzły robocze, z których każdy ma 8 vCPU i 32 GiB pamięci RAM:

Rysunek B. Przydział zasobów między węzłami roboczymi (utworzył Kuriko IWAI)

Łącznie w systemie dostępne są 24 vCPU.

Executory Sparka oraz alokacja CPU i pamięci

Załóżmy, że przydzielę:

• 7 executorów.
• 2 rdzenie vCPU dla każdego executora.
• 3 GiB pamięci sterty JVM dla każdego executora.

Aplikacja Spark zwraca się do menedżera klastra o przydzielenie zasobów tym 7 executorom.

Następnie menedżer klastra decyduje o przydziale procesorów i pamięci na podstawie wykorzystania zasobów przez poszczególne węzły robocze.

Na przykład decyduje o utworzeniu 14 slotów zadań — 2 rdzenie vCPU na executora × 7 executorów:

Rysunek C-1. Rdzenie vCPU, pamięć i przydział executorów między węzłami roboczymi przy nierównomiernym rozłożeniu zasobów w węźle roboczym A

Pamiętaj, że 3 GiB pamięci przydzielone executorowi nie stanowią całej jego pamięci, ponieważ trzeba zarezerwować dodatkową pamięć na potrzeby węzłów roboczych i innych zadań.

Zakładając, że pojedyncze zadanie wymaga 1 vCPU (spark.task.cpus = 1), każdy węzeł roboczy przydziela swoje vCPU:

Rysunek C-2. Równoległy przydział vCPU między węzłami roboczymi

Widać tu dwie istotne kwestie:

1. Wykorzystanie procesora w każdym węźle roboczym jest inne (na rysunku C-2 wykorzystanie w węźle roboczym A wynosi 75%, podczas gdy w węzłach B i C pozostaje na poziomie 50%).
2. Mimo że system ma 24 rdzenie vCPU (8 vCPU na węzeł roboczy × 3 węzły robocze), może wykorzystać tylko 14 rdzeni vCPU; pozostałe 10 pozostaje niewykorzystanych przy tych zadaniach.

2-3. Maszyna wirtualna Java (JVM) pod spodem Sparka

Apache Spark opiera się na maszynie wirtualnej Java (JVM).

JVM to środowisko uruchomieniowe wymagane do wykonywania kodu bajtowego Sparka.

Ponieważ Spark jest napisany w Scali i Javie:

• W przypadku użycia Scali lub Javy w aplikacji: kod działa natywnie w JVM.
• W przypadku użycia PySpark (Pythona): proces Pythona wykonuje kod, ale komunikuje się z sterownikiem opartym na JVM za pośrednictwem mostu, takiego jak Py4J. Spark executory w węzłach roboczych nadal działają w JVM, która następnie zarządza procesami roboczymi Pythona wykonującymi kod w Pythonie oraz funkcjami definiowanymi przez użytkownika w Pythonie (Python UDF).

W obu przypadkach JVM zarządza wątkami wykonania, przydziałem pamięci i odśmiecaniem pamięci zarówno dla sterownika, jak i executorów.

JVM sterownika & executorów w aplikacji PySpark

W aplikacji PySpark sterownik Pythona komunikuje się ze sterownikiem Spark działającym w JVM;

• Sterownik Pythona: tutaj zapisuje się i wykonuje kod PySpark na poziomie aplikacji.
• JVM sterownika: koordynuje wykonanie zadań Spark i ich planowanie.
• JVM executora: wykonuje zadania Spark i zarządza zasobami potrzebnymi do ich realizacji. Gdy zadania wymagają wykonania kodu w Pythonie, wykorzystywane są procesy robocze Pythona.
• JVM klienta (proces klienta lub maszyna, z której przesyłane jest zadanie): JVM działająca na komputerze lokalnym lub węźle brzegowym/bramowym.

Jak pokazano na rysunku, JVM klienta najpierw łączy się z JVM sterownika, która następnie koordynuje pracę z JVM-ami executorów:

Rysunek D. Ogólne zależności między JVM klienta, sterownika i executorów

Sterownik Spark działa w JVM sterownika i zarządza aplikacją, a w JVM-ach executorów działają poszczególne executory odpowiedzialne za wykonywanie tych zadań.

Każdy sterownik JVM ma stertę JVM, która umożliwia wydajne przetwarzanie i buforowanie danych oraz obsługę obiektów użytkownika.

2-4. Szczegółowe omówienie odpornych na awarie zbiorów rozproszonych (RDD)

Odporny na awarie zbiór rozproszony (RDD) to niezmienna kolekcja elementów podzielona na partycje, którą Spark może przetwarzać równolegle.

RDD śledzą zależności między transformacjami, dzięki czemu Spark może ponownie obliczyć utracone partycje w wielu scenariuszach awarii.

Nazwa ta odzwierciedla trzy właściwości:

• Odporny na awarie: Spark zapisuje pochodzenie danych — wskazuje, od którego wcześniejszego RDD zależy dany RDD — i często może odtworzyć brakujące partycje, zamiast przechowywać repliki każdego wyniku pośredniego.
• Rozproszony: zbiór danych jest reprezentowany przez partycje, które Spark może przetwarzać na różnych executorach. Nie są wymagane różne maszyny fizyczne; RDD są obsługiwane także w trybie lokalnym.
• Zbiór danych: RDD może zawierać pary klucz-wartość, liczby całkowite, ciągi znaków, krotki i inne obsługiwane elementy — nie tylko wiersze relacyjne. Interfejs API RDD obejmuje operacje takie jak map, filter, reduceByKey, join, oraz aggregateByKey.

RDD to abstrakcja logicznej kolekcji i obliczeń, a nie deklaracja, że określone partycje są na stałe przypisane do określonych procesów roboczych.

Partycje

Partycja to logiczny fragment danych, który zazwyczaj odpowiada jednemu zadaniu w ramach etapu.

Załóżmy, że tworzymy 100 partycji (wybór 100 partycji początkowych jest tu celowym uproszczeniem na potrzeby objaśnienia).

Przy 10 000 elementów źródłowych na każdą partycję przypada około 100 elementów, jeśli dane wejściowe są rozłożone w przybliżeniu równomiernie:

Rysunek E. Równoległy przydział partycji wśród 100 początkowych podziałów danych

3. API RDD a API DataFrame: wewnętrzna optymalizacja Catalyst

Pierwotny przykład z RDD pomaga zrozumieć pochodzenie danych, partycje i operację shuffle.

W typowych zadaniach ETL na danych ustrukturyzowanych najlepiej zacząć od API DataFrame, ponieważ udostępnia kolumny, informacje o schemacie, wyrażenia przypominające SQL oraz operacje widoczne dla optymalizatora.

Poniższy diagram przedstawia zależności między tymi elementami:

Rysunek F. Zależności między koncepcjami grafu w Spark: pochodzenie RDD, DataFrame i graf DAG wykonania w Spark

Podobnie jak pochodzenie RDD opisuje przekształcenia, plan zapytania DataFrame przedstawia jego ustrukturyzowane zapytanie.

Spark tworzy zadania do wykonania na podstawie zależności i operatorów fizycznych.

Wykonywanie operacji na DataFrame może wewnętrznie korzystać z mechanizmów opartych na RDD, ale nie oznacza to, że podczas pisania kodu DataFrame użytkownicy muszą ręcznie tworzyć pochodzenie RDD.

3-1. Plan zapytania DataFrame

Plan zapytania DataFrame to sekwencja kroków logicznych i fizycznych generowanych przez silnik wykonawczy (fioletowe pole na rysunku L).

Zapytanie DataFrame przechodzi przez kilka postaci:

1. Plan logiczny: opis żądanych operacji (skanowanie, filtrowanie, grupowanie, agregacja).
2. Przeanalizowany i zoptymalizowany plan logiczny: Spark rozpoznaje odwołania i stosuje dostępne optymalizacje logiczne, takie jak eliminowanie niepotrzebnych kolumn lub przesuwanie predykatów w dół planu.
3. Plan fizyczny: Spark wybiera operatory, które można wykonać, np. skanowanie, agregacje częściowe i końcowe, wymiany danych, złączenia i sortowanie.
4. Wykonanie: Plan fizyczny jest realizowany za pomocą mechanizmów wykonawczych Spark, a ostatecznie zadań uruchamianych na executorach.

W naszym przypadku plan fizyczny może obejmować następujące kroki:

• Częściowa agregacja haszująca,
• wymiana danych według SKU oraz
• końcowa agregacja haszująca.

3-2. Porównanie operacji: API RDD a API DataFrame

Porównajmy API RDD i API DataFrame na przykładzie agregacji według SKU.

API RDD

Elementy to (sku, qty) pary. Logika agregacji w Pythonie jest jawna, a Spark postępuje zgodnie z zależnościami 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

Załóżmy, że sales_df ma kolumny sku (ciąg znaków) i qty (liczbową):

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)]

W obu przykładach rozwiązujemy ten sam problem biznesowy:

Filtrowanie nieprawidłowych rekordów sprzedaży i agregowanie prawidłowych ilości według SKU.

W obu przypadkach collect() uruchamia wykonanie w Spark i zwraca wszystkie wynikowe rekordy do sterownika (Driver).

Kluczowa różnica dotyczy reprezentacji wyniku:

• API RDD: zwraca elementy Pythona, takie jak (sku, total_qty) krotki.
• API DataFrame: zwraca ustrukturyzowane Row obiekty z nazwanymi polami, takie jak sku i total_qty.

Poniżej przedstawiamy ogólne porównanie API RDD i DataFrame:

Tabela 2. Porównanie funkcji i poziomu abstrakcji: API RDD a API DataFrame (opracowała Kuriko IWAI)

4. Podsumowanie

Celem strojenia Sparka nie jest niezależne maksymalizowanie liczby partycji ani wykorzystania procesora.

Chodzi o zmniejszenie całkowitego kosztu i opóźnienia wykonania, bez wprowadzania niestabilności.

W tym artykule poznajemy podstawowe zależności w systemie Sparka:

• Zasoby: Węzły robocze hostują executory; executory uruchamiają zadania; skonfigurowane limity CPU i pamięci ograniczają współbieżność.
• Programowanie z użyciem RDD: Operacje transformacji tworzą niezmienne definicje RDD i informacje o ich pochodzeniu; akcje uruchamiają obliczenia.
• Programowanie z użyciem DataFrame: Operacje transformacji tworzą logiczne operacje zapytań; optymalizator wybiera fizyczne operatory, które można wykonać; akcje uruchamiają obliczenia.
• Wykonanie: Spark organizuje pracę w zadania, etapy i zadania wykonawcze, a zadania wykonawcze przetwarzają partycje w executorach. Tasowania danych i inne zależności tworzą granice koordynacji między etapami.

Zrozumienie, jak Spark wykonuje rozproszone obciążenia, to dopiero podstawa.

Możesz poznać siedmioetapowy przebieg obliczeń z użyciem API RDD.