Vad händer egentligen när du kör ett Spark-jobb?

Inledning

När datamängderna växer kan det bli svårt att bearbeta dem effektivt på en enda maskin på grund av begränsningar i processorkapacitet, minne och lagringsgenomströmning.

Apache Spark hanterar dessa utmaningar genom att tillhandahålla en distribuerad databehandlingsmotor som kan dela upp arbetsbelastningar i partitioner och köra uppgifter parallellt över processorkärnor och flera maskiner.

Spark handlar dock inte bara om att lägga till fler maskiner.

Det är viktigt att förstå hur komponenterna samverkar – driverprocesser, exekverare, RDD:er, DataFrames, partitioner, jobb, steg och uppgifter – för att utforma effektiva och tillförlitliga datapipelines.

I den här artikeln går jag igenom Sparks arkitektur, exekvering och viktiga prestandaaspekter med hjälp av ett praktiskt exempel på aggregering av e-handelsförsäljning.

1. Varför Apache Spark – parallell databehandling

Apache Spark är en distribuerad beräkningsmotor som är utformad för storskalig databehandling och datateknik.

Spark bearbetar stordata med hjälp av parallell databehandling:

Figur A. Översiktlig arkitektur för ett distribuerat Apache Spark-kluster (kluster-, driver- och arbetsnoder), med jämförelse med körning på en enda nod (skapad av Kuriko IWAI)

Jämfört med bearbetning på en enda nod (till höger i figur A) delar Spark upp arbetet i uppgifter med hjälp av flera noder och kör dem på tillgängliga resurser på olika maskiner.

Distribuerat betyder inte att flera maskiner är ett krav. Spark kan köras i lokalt läge på en enda maskin. Även då kan data partitioneras och bearbetas parallellt.

2. Apache Sparks grundläggande arkitektur

Spark består av flera viktiga komponenter:

• Spark-driver (drivrutin): Den centrala samordningsprocessen i en Spark-applikation.
• Klusterhanterare: En extern tjänst som ansvarar för att fördela maskinvaruresurser, till exempel processorkärnor och minne.
• Arbetsnoder (arbetare): En fysisk eller virtuell maskin eller beräkningsnod i klustret.
• Exekverare: Distribuerade arbetsprocesser som startas på arbetsnoderna.
• Java Virtual Machine (JVM): Körmiljön som krävs för att exekvera Sparks bytekod (eftersom Spark är skrivet i Scala/Java).

Först samordnar drivrutinen en Spark-applikation och omvandlar den till en riktad acyklisk graf (DAG) med fysiska exekveringsuppgifter, som sedan schemaläggs på exekverarna.

Därefter fördelar klusterhanteraren resurser enligt driftsättningens resurshanteringssystem.

Arbetsnoderna tillhandahåller beräkningsresurser åt exekverarna, som kör uppgifter och lagrar data medan uppgifterna körs.

Vissa hanterade miljöer, till exempel Databricks, använder en exekverare per arbetsnod, men det är ingen generell regel i Spark. Beroende på klusterkonfigurationen kan en arbetsnod vara värd för flera exekverare.

2-1. E-handelsfall – indatastruktur

Föreställ dig ett litet e-handelsföretag.

Företaget har 10 000 försäljningsposter, som var och en innehåller en lagerföringsenhet (SKU) och en såld kvantitet. Anta att det finns tre möjliga SKU:er: A, B, och C.

Tabell 1. Exempel på fördelning av försäljningsposter inom e-handel och förväntade aggregeringar.

Målet är att beräkna det totala antalet giltiga sålda enheter per SKU:

1. Filtrera bort poster vars qty <= 0, och sedan
2. Aggregera posterna per SKU.

I det illustrativa fullständiga datasetet antar vi att 9 800 av de 10 000 posterna återstår efter filtreringen. De slutliga summor som förväntas från Spark är:

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

2-2. Spark-klustret

Jag definierar ett förenklat kluster för vårt användningsfall.

Arbetsnoder (workers)

Jag använder 3 arbetsnoder, var och en med en instanstyp på 8 vCPU och 32 GiB RAM:

Figur B. Resursfördelning mellan arbetsnoder (skapad av Kuriko IWAI)

Totalt finns 24 vCPU tillgängliga i systemet.

Spark-exekverare och tilldelning av CPU och minne

Låt oss säga att jag reserverar:

• 7 exekverare.
• 2 vCPU-kärnor per exekverare.
• 3 GiB JVM-heapminne per exekverare.

Spark-applikationen begär att klusterhanteraren tilldelar resurser till dessa 7 exekverare.

Därefter avgör klusterhanteraren hur processorer och minne ska fördelas utifrån resursanvändningen på varje arbetarnod.

Den kan till exempel besluta att skapa 14 aktivitetsplatser – 2 vCPU-kärnor per exekverare × 7 exekverare:

Figur C-1. vCPU-kärnor, minne och fördelning av exekverare mellan arbetarnoder vid ojämn resursfördelning på arbetarnod A

Observera att de 3 GiB minne som tilldelas exekveraren inte utgör dess totala minne, eftersom den måste reservera minne för arbetarnodernas omkostnader och annan körning.

Om vi antar att en enskild aktivitet kräver 1 vCPU (spark.task.cpus = 1), fördelar varje arbetarnod sina vCPU:er:

Figur C-2. Parallell fördelning av vCPU mellan arbetarnoder

Det här visar två viktiga saker:

1. Varje arbetarnod har en egen CPU-användningsgrad (i figur C-2 är användningen på arbetarnod A 75 %, medan arbetarnod B och C ligger kvar på 50 %).
2. Även om systemet har 24 vCPU-kärnor (8 vCPU per arbetarnod × 3 arbetarnoder) kan det bara använda 14 vCPU-kärnor; de återstående 10 används inte för dessa aktiviteter.

2-3. Java Virtual Machine (JVM) bakom Spark

Apache Spark bygger på Java Virtual Machine (JVM).

JVM är en körmiljö som krävs för att köra Sparks bytekod.

Eftersom Spark är skrivet i Scala/Java:

• När Scala eller Java används i applikationen: Körs koden nativt i JVM.
• När PySpark (Python) används: Kör en Python-process koden, men kommunicerar med en JVM-baserad drivrutin via en brygga som Py4J. Spark-exekverarna på arbetarnoderna körs fortfarande i en JVM, som sedan hanterar Python-arbetarprocesser för att köra Python-kod och Python-användardefinierade funktioner (Python UDF:er).

I båda fallen hanterar JVM körningstrådar, minnesfördelning och skräpinsamling för både drivrutinen och exekverarna.

JVM:er för drivrutin & exekverare i PySpark-applikationer

I en PySpark-applikation kommunicerar Python-drivrutinen med Spark-drivrutinen som körs i JVM:

• Python-drivrutin: Här skrivs och körs PySpark-koden på applikationsnivå.
• JVM-drivrutin: Samordnar Spark-körningen och schemaläggningen.
• JVM-exekverare: Kör Spark-aktiviteter och hanterar körningsresurser. Python-arbetarprocesser används när aktiviteter kräver Python-körning.
• Klient-JVM (klientprocess eller inlämningsmaskin): JVM som körs på den lokala maskinen eller en kantnod/gatewaynod.

Som figuren visar ansluter klient-JVM:en först till JVM-drivrutinen, som därefter samordnar med JVM-exekverarna:

Figur D. Övergripande relation mellan klientens, drivrutinens och exekverarnas JVM

Därefter är Spark-drivrutinen värd i JVM-drivrutinen och hanterar applikationen, medan de enskilda exekverarna som ansvarar för att utföra aktiviteterna finns i JVM-exekverarna.

Varje JVM-drivrutin har en JVM-heap för effektiv hantering av databearbetning, cachelagring och användarobjekt.

2-4. Fördjupning i Resilient Distributed Datasets (RDD)

En feltålig distribuerad datamängd (RDD) är en oföränderlig, partitionerad samling element som Spark kan bearbeta parallellt.

RDD:er håller reda på beroenden mellan transformationer, vilket gör att Spark kan återskapa förlorade partitioner i många typer av felsituationer.

Namnet beskriver tre egenskaper:

• Feltålig: Spark registrerar härstamningen – det vill säga vilken RDD som är beroende av vilken tidigare RDD – och kan ofta återskapa saknade partitioner i stället för att behålla repliker av varje mellanresultat.
• Distribuerad: Datamängden representeras av partitioner som Spark kan bearbeta på olika exekverare. Flera fysiska maskiner krävs inte; RDD:er stöds även i lokalt läge.
• Datamängd: En RDD kan innehålla nyckel-värdepar, heltal, strängar, tupler och andra element som stöds – inte bara relationsrader. RDD-API:erna innehåller operationer som map, filter, reduceByKey, join, och aggregateByKey.

En RDD är en logisk samlings- och beräkningsabstraktion, inte en deklaration om att vissa partitioner permanent tilldelas vissa arbetare.

Partitioner

En partition är en logisk del av data och motsvarar vanligtvis en uppgift inom ett steg.

Låt oss säga att vi skapar 100 partitioner (dessa 100 initiala partitioner är ett uttryckligt pedagogiskt val).

Med 10 000 källelement blir det ungefär 100 element per partition om indata fördelas någorlunda jämnt:

Figur E. Parallell fördelning av partitioner över 100 initiala datadelningar

3. RDD API jämfört med DataFrame API: intern optimering med Catalyst

Det ursprungliga RDD-exemplet hjälper oss att förstå härkomst, partitioner och shuffle.

I typiska strukturerade ETL-flöden är det dock DataFrame API som är den rekommenderade utgångspunkten, eftersom det tillhandahåller kolumner, schemainformation, SQL-liknande uttryck och operationer som optimeraren kan ta hänsyn till.

Diagrammet nedan visar sambandet:

Figur F. Sambandet mellan Sparks grafbegrepp: RDD-härkomst, DataFrame och Sparks exekverings-DAG

Precis som RDD har RDD-härkomst som definierar transformationen, representerar DataFrames frågeplan den strukturerade frågan.

Spark skapar exekveringsarbete utifrån beroenden och fysiska operatorer.

DataFrame-exekvering kan använda RDD-baserade mekanismer internt, men det betyder inte att användare måste skapa RDD-härkomst manuellt när de skriver DataFrame-kod.

3-1. DataFrames frågeplan

En DataFrame-frågeplan är den sekvens av logiska och fysiska steg som en exekveringsmotor genererar (den lila rutan i figur L).

En DataFrame-fråga passerar genom flera representationer:

1. Logisk plan: En beskrivning av de begärda operationerna (genomsökning, filtrering, gruppering, aggregering).
2. Analyserad och optimerad logisk plan: Spark löser referenser och tillämpar relevanta logiska optimeringar, till exempel kolumnbeskärning eller predikatpushning.
3. Fysisk plan: Spark väljer körbara operatorer, till exempel genomsökningar, partiella/slutliga aggregeringar, datautbyten, kopplingar och sorteringar.
4. Exekvering: Den fysiska planen körs med Sparks exekveringsmekanismer och omfattar i slutändan uppgifter på exekverare.

I vårt användningsfall kan den fysiska planen innehålla följande steg:

• Partiell hashaggregering,
• datautbyte efter SKU, och
• slutlig hashaggregering.

3-2. Jämförelse av åtgärder: RDD API jämfört med DataFrame API

Låt oss jämföra RDD API och DataFrame API med aggregering efter SKU som exempel.

RDD API

Elementen är (sku, qty) par. Vår aggregeringslogik i Python är explicit, och Spark följer RDD-beroenden:

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

DataFrame API

Anta att sales_df har kolumnerna sku (sträng) och qty (numerisk):

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

Här löser båda exemplen samma affärsproblem:

Filtrering av ogiltiga försäljningsposter och aggregering av giltiga kvantiteter efter SKU.

Och i båda fallen collect() utlöser Spark-exekvering och returnerar alla resulterande poster till drivrutinen.

Den viktigaste skillnaden här gäller hur resultatet representeras:

• RDD API: Returnerar Python-element, till exempel (sku, total_qty) tupler.
• DataFrame API: Returnerar strukturerade Row objekt med namngivna fält, till exempel sku och total_qty.

Här är en övergripande jämförelse av RDD API och DataFrame API:

Tabell 2. RDD API jämfört med DataFrame API – jämförelse av funktioner och abstraktionsnivå: (Skapad av Kuriko IWAI)

4. Sammanfattning

Målet med Spark-optimering är inte att maximera antalet partitioner eller CPU-användningen var för sig.

Målet är att minska den totala exekveringskostnaden och fördröjningen utan att skapa instabilitet.

I den här artikeln går vi igenom de grundläggande sambanden i systemet:

• Resurser: Arbetarnoder är värdar för exekverare; exekverare kör uppgifter; konfigurerad CPU och minne begränsar samtidigheten.
• RDD-programmering: Transformationer skapar oföränderliga RDD-definitioner och en härkomstkedja; åtgärder utlöser beräkning.
• DataFrame-programmering: Transformationer skapar logiska frågeoperationer; optimeringen väljer fysiska operatorer som kan köras; åtgärder utlöser beräkning.
• Exekvering: Spark organiserar arbetet i jobb, steg och uppgifter, och uppgifterna bearbetar partitioner på exekverare. Shufflar och andra beroenden skapar samordningsgränser mellan stegen.

Att förstå hur Spark kör distribuerade arbetslaster är bara grunden.

Du kan lära dig ett beräkningsflöde i sju steg med RDD API.