Co se vlastně děje, když spustíte úlohu Spark?

Úvod

S rostoucími objemy dat může být jejich efektivní zpracování na jediném počítači náročné kvůli omezené kapacitě procesoru, paměti a propustnosti úložiště.

Apache Spark tyto problémy řeší tím, že nabízí distribuovaný engine pro zpracování dat, který dokáže rozdělit pracovní zátěž na oddíly a spouštět úlohy paralelně napříč jádry procesoru i na více počítačích.

Spark však neznamená jen přidání dalších počítačů.

Pro návrh efektivních a spolehlivých datových pipeline je zásadní pochopit, jak spolupracují jeho komponenty — ovladače, exekutory, RDD, DataFrames, oddíly, úlohy, fáze a dílčí úlohy.

V tomto článku se na praktickém příkladu agregace prodejů v e-commerce podívám na architekturu Sparku, mechanismy jeho vykonávání a klíčové faktory ovlivňující výkon.

1. Proč Apache Spark – paralelní zpracování dat

Apache Spark je distribuovaný výpočetní engine navržený pro rozsáhlé zpracování dat a datové inženýrství.

Spark zpracovává velká data pomocí paralelního zpracování:

Obrázek A. Přehled architektury distribuovaného clusteru Apache Spark (cluster, ovladač a pracovní uzly) se srovnáním se zpracováním na jediném uzlu (vytvořila Kuriko IWAI)

Na rozdíl od zpracování na jediném uzlu (vpravo, obrázek A) Spark rozděluje práci na úlohy pomocí více uzlů a vykonává je na dostupných prostředcích napříč počítači.

Distribuované zpracování zde neznamená, že je nutné použít více počítačů. Spark může běžet v lokálním režimu na jediném počítači. I v takovém případě lze data rozdělit na oddíly a zpracovávat paralelně.

2. Základní architektura Apache Spark

Spark se skládá z několika klíčových komponent:

• Ovladač Sparku (driver): Centrální proces, který koordinuje aplikaci Spark.
• Správce clusteru: Externí služba zodpovědná za přidělování hardwarových prostředků, například jader procesoru a paměti.
• Pracovní uzly (pracovní uzel): Fyzický nebo virtuální počítač, případně výpočetní uzel v clusteru.
• Exekutory: Distribuované pracovní procesy spouštěné na pracovních uzlech.
• Java Virtual Machine (JVM): Běhové prostředí potřebné ke spuštění bajtkódu Sparku (protože Spark je napsán v jazycích Scala a Java).

Nejprve ovladač koordinuje aplikaci Spark a převádí ji na orientovaný acyklický graf (DAG) fyzických úloh, jejichž vykonávání plánuje napříč exekutory.

Poté správce clusteru přiděluje prostředky v rámci systému správy prostředků daného nasazení.

Pracovní uzly poskytují exekutorům výpočetní prostředky; exekutoři pak vykonávají úlohy a během jejich provádění uchovávají data.

Některá spravovaná prostředí, například Databricks, používají jeden exekutor na pracovní uzel, to však není univerzálním pravidlem Sparku. V závislosti na konfiguraci clusteru může jeden pracovní uzel hostovat více exekutorů.

2–1. Příklad z e-commerce – struktura vstupních dat

Představme si malou e-commerce společnost.

Má 10 000 záznamů o prodejích, z nichž každý obsahuje skladovou jednotku (SKU) a prodané množství. Předpokládejme, že jsou možné tři SKU: A, B, a C.

Tabulka 1. Příklad rozdělení záznamů o prodejích v e-commerce a očekávaných agregací.

Naším cílem je vypočítat celkový počet platných prodaných kusů pro každé SKU:

1. Vyřadit záznamy, jejichž qty <= 0, potom
2. Agregovat záznamy podle SKU.

Předpokládejme, že v tomto ilustrativním úplném souboru dat po filtrování zůstane 9 800 z 10 000 záznamů. Očekávané konečné součty ze Sparku jsou:

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

2–2. Cluster Spark

Pro náš příklad nastavím zjednodušený cluster.

Pracovní uzly (workeři)

Nastavím 3 pracovní uzly, každý s typem instance 8 vCPU a 32 GiB RAM:

Obrázek B. Přidělení prostředků napříč pracovními uzly (vytvořila Kuriko IWAI)

Systém má celkem k dispozici 24 vCPU.

Spark executory a přidělování procesorů a paměti

Řekněme, že si zajistím:

• 7 executorů.
• 2 jádra vCPU pro každý executor.
• 3 GiB paměti haldy JVM pro každý executor.

Aplikace Spark požádá správce clusteru o přidělení prostředků těmto 7 executorům.

Správce clusteru poté rozhodne o přidělení procesorů a paměti podle využití prostředků jednotlivými pracovními uzly.

Například se rozhodne vytvořit 14 slotů pro úlohy – 2 jádra vCPU na executor × 7 executorů:

Obrázek C-1. Jádra vCPU, paměť a přidělení executorů mezi pracovními uzly při nerovnoměrném rozložení prostředků na pracovním uzlu A

Upozorňujeme, že 3 GiB paměti přidělené executorům nepředstavují celkové množství jejich paměti, protože je třeba zajistit také režii paměti pro pracovní uzly a další procesy.

Předpokládejme, že jedna úloha potřebuje 1 vCPU (spark.task.cpus = 1); každý pracovní uzel přiděluje svá jádra vCPU takto:

Obrázek C-2. Paralelní přidělování jader vCPU mezi pracovními uzly

Z toho vyplývají dva důležité poznatky:

1. Každý pracovní uzel má jedinečnou míru využití procesoru (na obrázku C-2 je využití pracovního uzlu A 75 %, zatímco u pracovních uzlů B a C zůstává na 50 %).
2. Ačkoli má systém 24 jader vCPU (8 vCPU na pracovní uzel × 3 pracovní uzly), může využít pouze 14 jader vCPU; zbývajících 10 zůstává pro tyto úlohy nevyužito.

2-3. Java Virtual Machine (JVM) v zákulisí Sparku

Apache Spark je postaven na platformě Java Virtual Machine (JVM).

JVM je běhové prostředí potřebné ke spuštění bajtkódu Sparku.

Protože je Spark napsán ve Scale a Javě:

• Při použití Scaly nebo Javy v aplikaci: Kód běží nativně uvnitř JVM.
• Při použití PySparku (Pythonu): Kód spouští proces Pythonu, který komunikuje s ovladačem založeným na JVM prostřednictvím mostu, jako je Py4J. Exekutory Sparku na pracovních uzlech stále běží uvnitř JVM, který následně spravuje pracovní procesy Pythonu pro spouštění kódu Pythonu a uživatelsky definovaných funkcí Pythonu (Python UDF).

V obou případech JVM spravuje vlákna vykonávání, přidělování paměti a uvolňování nepoužívané paměti pro ovladač i executory.

JVM ovladače a executorů v aplikaci PySpark

V aplikaci PySpark komunikuje ovladač Pythonu s ovladačem Sparku běžícím v JVM;

• Ovladač Pythonu: Zde se píše a spouští kód aplikace PySpark.
• JVM ovladače: Koordinuje vykonávání Sparku a plánování úloh.
• JVM executorů: Spouští úlohy Sparku a spravuje prostředky potřebné k jejich vykonávání. Pokud úlohy vyžadují spuštění kódu Pythonu, zapojují se pracovní procesy Pythonu.
• JVM klienta (klientský proces nebo počítač, ze kterého se aplikace odesílá): JVM běžící na místním počítači nebo na hraničním či přístupovém uzlu.

Jak ukazuje obrázek, JVM klienta se nejprve připojí k JVM ovladače, který následně koordinuje činnost JVM executorů:

Obrázek D. Celkové vztahy mezi JVM klienta, ovladače a executorů

JVM ovladače zde hostuje ovladač Sparku, který řídí aplikaci, zatímco JVM executorů hostují jednotlivé executory odpovědné za vykonávání těchto úloh.

Každý ovladač JVM má vlastní haldu JVM, která efektivně zpracovává data, ukládá je do mezipaměti a spravuje uživatelské objekty.

2-4. Podrobný přehled odolných distribuovaných datových sad (RDD)

Odolná distribuovaná datová sada (RDD) je neměnná, rozdělená kolekce prvků, kterou může Spark zpracovávat paralelně.

RDD sledují závislosti mezi transformacemi, což Sparku umožňuje v mnoha případech selhání znovu vypočítat ztracené oddíly.

Tento název vystihuje tři vlastnosti:

• Odolná: Spark zaznamenává historii závislostí – tedy které RDD závisí na kterém předchozím RDD – a často dokáže chybějící oddíly znovu vytvořit, místo aby uchovával repliky každého mezivýsledku.
• Distribuovaná: Datová sada je reprezentována oddíly, které může Spark zpracovávat na různých executorech. Více fyzických počítačů není nutných; RDD podporuje také místní režim.
• Datová sada: RDD může obsahovat páry klíč–hodnota, celá čísla, řetězce, n-tice a další podporované prvky – nejen relační řádky. Rozhraní API pro RDD zahrnují operace jako map, filter, reduceByKey, join, a aggregateByKey.

RDD je abstrakce logické kolekce a výpočtu, nikoli deklarace, že konkrétní oddíly jsou trvale přiřazeny konkrétním pracovním uzlům.

Oddíly

Oddíl je logický blok dat a obvykle odpovídá jedné úloze v rámci fáze.

Řekněme, že vytvoříme 100 oddílů (těchto 100 počátečních oddílů jsme zvolili výslovně pro názornost).

Při 10 000 zdrojových prvcích připadá na každý oddíl přibližně 100 prvků, pokud jsou vstupní data rozložena zhruba rovnoměrně:

Obrázek E. Paralelní přiřazení oddílů napříč 100 počátečními rozděleními dat

3. API RDD vs. API DataFrame: interní optimalizace Catalyst

Původní příklad s RDD pomáhá porozumět lineage, oddílům a operaci shuffle.

V typickém strukturovaném ETL je však API DataFrame vhodnějším výchozím bodem, protože nabízí sloupce, informace o schématu, výrazy podobné SQL a operace viditelné pro optimalizátor.

Následující diagram znázorňuje tento vztah:

Obrázek F. Vztah mezi koncepty grafů ve Sparku: lineage RDD, DataFrame a DAG vykonávání Sparku

Stejně jako lineage RDD popisuje transformace, plán dotazu DataFrame představuje jeho strukturovaný dotaz.

Spark sestavuje úlohy vykonávání na základě závislostí a fyzických operátorů.

Při vykonávání DataFrame se mohou interně používat mechanismy založené na RDD, ale to neznamená, že uživatelé musí při psaní kódu DataFrame ručně vytvářet lineage RDD.

3–1. Plán dotazu DataFrame

Plán dotazu DataFrame je posloupnost logických a fyzických kroků, kterou vytvoří vykonávací engine (fialový rámeček na obrázku L).

Dotaz DataFrame prochází několika reprezentacemi:

1. Logický plán: Popis požadovaných operací (načtení, filtrování, seskupení, agregace).
2. Analyzovaný a optimalizovaný logický plán: Spark vyřeší odkazy a použije příslušné logické optimalizace, například odstranění nepotřebných sloupců nebo posunutí predikátu.
3. Fyzický plán: Spark zvolí vykonatelné operátory, například načítání, částečné a konečné agregace, přenosy dat, spojení a řazení.
4. Vykonávání: Fyzický plán se spustí pomocí vykonávacích mechanismů Sparku a nakonec zahrnuje úlohy na executorech.

V našem případě použití může fyzický plán zahrnovat tyto kroky:

• Částečná hashovací agregace,
• Přerozdělení podle SKU a
• Konečná hashovací agregace.

3–2. Porovnání akcí: API RDD vs. API DataFrame

Porovnejme API RDD a API DataFrame na příkladu agregace podle SKU.

API RDD

Prvky jsou (sku, qty) dvojice. Naše agregační logika v Pythonu je explicitní a Spark postupuje podle závislostí 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

Předpokládejme, že sales_df má sloupce sku (řetězec) a qty (číselný):

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

V obou příkladech řešíme stejný obchodní problém:

Filtrování neplatných záznamů o prodeji a agregace platných množství podle SKU.

A v obou případech collect() spustí vykonávání ve Sparku a vrátí všechny výsledné záznamy ovladači (Driver).

Hlavní rozdíl spočívá v reprezentaci výsledku:

• API RDD: Vrací prvky Pythonu, například (sku, total_qty) n-tice.
• API DataFrame: Vrací strukturované Row objekty s pojmenovanými poli, například sku a total_qty.

Celkové srovnání API RDD a DataFrame:

Tabulka 2. Porovnání funkcí a úrovně abstrakce rozhraní RDD API a DataFrame API: (vytvořila Kuriko IWAI)

4. Závěrem

Cílem ladění Sparku není nezávisle maximalizovat počet oddílů ani využití procesoru.

Jde o snížení celkových nákladů na vykonávání a latence, aniž by se narušila stabilita.

V tomto článku se seznámíme se základními souvislostmi v jeho systému:

• Prostředky: Pracovní uzly hostí exekutory; exekutory spouštějí úkoly; nastavené limity procesoru a paměti omezují souběžnost.
• Programování pomocí RDD: Transformace vytvářejí neměnné definice RDD a posloupnost jejich závislostí; akce spouštějí výpočet.
• Programování pomocí DataFrame: Transformace vytvářejí logické operace dotazu; optimalizace vybírá fyzické operátory, které lze vykonat; akce spouštějí výpočet.
• Provádění: Spark organizuje práci do úloh, fází a úkolů a úkoly zpracovávají oddíly v exekutorech. Přesuny dat a další závislosti vytvářejí hranice pro koordinaci mezi fázemi.

Porozumění tomu, jak Spark vykonává distribuované úlohy, je teprve základem.

Můžete se seznámit se sedmikrokovým průběhem výpočtu pomocí rozhraní RDD API.