Hva skjer egentlig når du kjører en Spark-jobb?

Innledning

Etter hvert som datasett vokser, kan det bli krevende å behandle dem effektivt på én enkelt maskin på grunn av begrensninger i prosessorkapasitet, minne og lagringsgjennomstrømning.

Apache Spark løser disse utfordringene ved å tilby en distribuert databehandlingsmotor som kan dele arbeidsmengder inn i partisjoner og utføre oppgaver parallelt på tvers av CPU-kjerner og flere maskiner.

Spark handler imidlertid ikke bare om å legge til flere maskiner.

Det er avgjørende å forstå hvordan komponentene fungerer sammen – drivere, executor-prosesser, RDD-er, DataFrames, partisjoner, jobber, trinn og oppgaver – for å utforme effektive og pålitelige datapipelines.

I denne artikkelen går jeg gjennom Sparks arkitektur, hvordan kjøringen foregår og viktige hensyn knyttet til ytelse, ved hjelp av et praktisk eksempel på aggregering av e-handelssalg.

1. Hvorfor Apache Spark – parallell databehandling

Apache Spark er en distribuert databehandlingsmotor utviklet for storskala databehandling og datateknikk.

Spark behandler stordata ved hjelp av parallell databehandling:

Figur A. Overordnet arkitektur for en distribuert Apache Spark-klynge (klynge, driver- og arbeidernoder), med sammenligning av kjøring på én enkelt node (laget av Kuriko IWAI)

Sammenlignet med behandling på én enkelt node (til høyre i Figur A) deler Spark arbeidet inn i oppgaver som kjøres på flere noder, og utfører dem på tvers av tilgjengelige ressurser på tvers av maskiner.

Distribuert betyr ikke at flere maskiner er påkrevd. Spark kan kjøres lokalt på én enkelt maskin. Også da kan data deles inn i partisjoner og behandles parallelt.

2. Kjernekomponenter i Apache Sparks arkitektur

Spark består av flere sentrale komponenter:

• Spark-driver (driver): Den sentrale prosessen som koordinerer Spark-applikasjonen.
• Klyngeadministrator: En ekstern tjeneste som har ansvar for å tildele maskinvareressurser, for eksempel CPU-kjerner og minne.
• Arbeidernoder (arbeider): En fysisk eller virtuell maskin eller en beregningsnode i klyngen.
• Executor-prosesser: Distribuerte arbeiderprosesser som startes på arbeidernodene.
• Java Virtual Machine (JVM): Kjøremiljøet som kreves for å kjøre Spark-bytekode (ettersom Spark er skrevet i Scala/Java).

Først koordinerer driveren en Spark-applikasjon og omgjør den til en rettet asyklisk graf (DAG) av fysiske kjøreoppgaver, som den planlegger og fordeler på executor-prosessene.

Deretter tildeler klyngeadministratoren ressurser i henhold til ressursstyringssystemet for utrullingen.

Arbeidernodene stiller beregningsressurser til rådighet for executor-prosessene, som utfører oppgaver og lagrer data mens oppgavene kjøres.

Noen administrerte miljøer, som Databricks, bruker én executor-prosess per arbeider, men dette er ingen generell Spark-regel. En arbeider kan kjøre flere executor-prosesser, avhengig av klyngekonfigurasjonen.

2-1. Brukstilfelle innen e-handel – datastruktur for inndata

La oss se for oss et lite e-handelsselskap.

Det har 10,000 salgsposter, hver med en lagerføringsenhet (SKU) og et solgt antall. Anta at det finnes tre mulige SKU-er: A, B, og C.

Tabell 1. Eksempel på fordeling av e-handelssalgsposter og forventede aggregeringer.

Målet vårt er å beregne totalt antall gyldige solgte enheter per SKU:

1. Filtrer bort poster der qty <= 0, og deretter
2. Aggreger postene etter SKU.

I det illustrerende, fullstendige datasettet kan vi anta at 9,800 av de 10,000 postene gjenstår etter filtreringen. De endelige summene som forventes fra Spark, er:

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

2-2. Spark-klynge

Jeg definerer en forenklet klynge for brukstilfellet vårt.

Arbeidernoder (arbeidere)

Jeg setter opp 3 arbeidernoder, som hver har en instanstype med 8 vCPU-er og 32 GiB RAM:

Figur B. Ressursfordeling på tvers av arbeidernoder (laget av Kuriko IWAI)

Totalt er 24 vCPU-er tilgjengelige i systemet.

Spark-executorer og tildeling av CPU og minne

La oss si at jeg sikrer meg:

• 7 executorer.
• 2 vCPU-kjerner for hver executor.
• 3 GiB JVM-heapminne for hver executor.

Spark-applikasjonen ber klyngebehandleren om å tildele ressurser til disse 7 executorene.

Deretter avgjør klyngebehandleren hvordan CPU-er og minne skal fordeles, basert på ressursbruken til hver worker.

For eksempel kan den opprette 14 oppgaveplasser – 2 vCPU-kjerner per executor × 7 executorer:

Figur C-1. vCPU-kjerner, minne og executor-tildeling på tvers av workere, med ubalansert ressursbruk på Worker A

Merk at de 3 GiB med minne som er tildelt executoren, ikke er dens totale minne, fordi den må reservere minneoverhead til workere og andre kjøringer.

Forutsatt at én enkelt oppgave krever 1 vCPU (spark.task.cpus = 1) tildeler hver worker sine vCPU-er:

Figur C-2. Parallell tildeling av vCPU-er på tvers av workere

Dette viser to viktige poenger:

1. Hver worker har sin egen CPU-bruksgrad (i Figur C-2 er bruken på Worker A 75 %, mens Worker B og C fortsatt ligger på 50 %).
2. Selv om systemet har 24 vCPU-kjerner (8 vCPU-er per worker × 3 workere), kan det bare bruke 14 vCPU-kjerner; de resterende 10 blir ikke brukt til disse oppgavene.

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

Apache Spark er bygget på Java Virtual Machine (JVM).

JVM er et kjøretidsmiljø som kreves for å kjøre Sparks bytekode.

Siden Spark er skrevet i Scala/Java:

• Når du bruker Scala eller Java i applikasjonen: Koden kjøres direkte i JVM.
• Når du bruker PySpark (Python): En Python-prosess kjører koden, men kommuniserer med en JVM-basert driver via en bro, for eksempel Py4J. Spark-executorene på workernodene kjører fortsatt i en JVM, som deretter administrerer Python-workere for å kjøre Python-kode og Python-brukerdefinerte funksjoner (Python UDF-er).

I begge tilfeller administrerer JVM kjøringstråder, minnetildeling og søppelrydding for både driveren og executorene.

Driver- og executor-JVM-er i en PySpark-applikasjon

I en PySpark-applikasjon kommuniserer Python-driveren med Spark-driveren som kjører i JVM:

• Python-driver: Her skrives og kjøres PySpark-kode på applikasjonsnivå.
• Driver-JVM: Koordinerer Spark-kjøringen og planleggingen.
• Executor-JVM: Kjører Spark-oppgaver og administrerer kjøringsressurser. Python-arbeiderprosesser brukes når oppgavene krever kjøring av Python-kode.
• Klient-JVM (klientprosess eller maskinen som sender inn jobben): JVM som kjører på den lokale maskinen eller en edge-/gateway-node.

Som vist i figuren kobler klient-JVM-en seg først til driver-JVM-en, som deretter koordinerer med executor-JVM-ene:

Figur D. Overordnet sammenheng mellom klient-, driver- og executor-JVM-er

Herfra er Spark-driveren vert på driver-JVM-en og administrerer applikasjonen, mens de enkelte executorene som er ansvarlige for å kjøre oppgavene, er vert på executor-JVM-ene.

Hver driver-JVM har en JVM-heap som effektivt håndterer databehandling, mellomlagring og brukerobjekter.

2-4. Grundig gjennomgang av Resilient Distributed Datasets (RDD-er)

En Resilient Distributed Dataset (RDD) er en uforanderlig, partisjonert samling av elementer som Spark kan behandle parallelt.

RDD-er holder oversikt over avhengigheter mellom transformasjoner, slik at Spark kan beregne tapte partisjoner på nytt i mange typer feilscenarier.

Navnet viser til tre egenskaper:

• Robust: Spark registrerer avstamningen – hvilken RDD som avhenger av hvilken tidligere RDD – og kan ofte gjenopprette manglende partisjoner i stedet for å opprettholde replikaer av hvert mellomresultat.
• Distribuert: Datasettet er representert som partisjoner som Spark kan behandle på ulike executorer. Det er ikke nødvendig med flere fysiske maskiner; RDD-er støttes også i lokal modus.
• Datasett: En RDD kan inneholde nøkkel-verdi-par, heltall, strenger, tupler og andre støttede elementer – ikke bare relasjonelle rader. RDD-API-ene omfatter operasjoner som map, filter, reduceByKey, join, og aggregateByKey.

En RDD er en logisk samling og beregningsabstraksjon, ikke en erklæring om at bestemte partisjoner er permanent tilordnet bestemte arbeidere.

Partisjoner

En partisjon er en logisk del av dataene og tilsvarer vanligvis én oppgave i et trinn.

La oss si at vi oppretter 100 partisjoner (disse 100 opprinnelige partisjonene er et bevisst valg for å gjøre eksempelet pedagogisk).

Med 10 000 kildedataelementer gir det omtrent 100 elementer per partisjon hvis inndataene er fordelt noenlunde jevnt:

Figur E. Parallell fordeling av partisjoner på 100 opprinnelige datadeler

3. RDD-API kontra DataFrame-API: intern Catalyst-optimalisering

Det opprinnelige RDD-eksempelet gjør det lettere å forstå avstamning, partisjoner og omfordeling.

I vanlig strukturert ETL er DataFrame-API-et likevel det foretrukne utgangspunktet, fordi det gir kolonner, skjemainformasjon, SQL-lignende uttrykk og operasjoner som optimalisereren kan se.

Diagrammet nedenfor viser sammenhengen:

Figur F. Sammenhengen mellom Spark-grafens begreper: RDD-avstamning, DataFrame og Sparks kjørings-DAG

På samme måte som en RDD har en RDD-avstamning som beskriver transformasjonene, representerer DataFrames spørringsplan den strukturerte spørringen.

Spark bygger kjøringsarbeidet ut fra avhengigheter og fysiske operatorer.

DataFrame-kjøring kan bruke RDD-baserte mekanismer internt, men det betyr ikke at brukerne må bygge RDD-avstamningen manuelt når de skriver DataFrame-kode.

3-1. DataFrames spørringsplan

En DataFrame-spørringsplan er sekvensen av logiske og fysiske trinn som en kjøringsmotor genererer (den lilla boksen i figur L).

En DataFrame-spørring går gjennom flere representasjoner:

1. Logisk plan: En beskrivelse av de forespurte operasjonene (skann, filtrering, gruppering, aggregering).
2. Analysert og optimalisert logisk plan: Spark løser referanser og bruker relevante logiske optimaliseringer, som kolonnebeskjæring eller predikatnedskyving.
3. Fysisk plan: Spark velger kjørbare operatorer, for eksempel skanning, delvise og endelige aggregeringer, utvekslinger, sammenføyninger og sorteringer.
4. Kjøring: Den fysiske planen kjøres ved hjelp av Sparks kjøremekanismer, og omfatter til slutt oppgaver på executor-noder.

I vårt bruksområde kan den fysiske planen omfatte følgende trinn:

• Delvis hash-aggregering,
• utveksling etter SKU, og
• endelig hash-aggregering.

3-2. Sammenligning av operasjoner: RDD-API kontra DataFrame-API

La oss sammenligne RDD-API-et og DataFrame-API-et med aggregering etter SKU som eksempel.

RDD-API

Elementene er (sku, qty) par. Aggregeringslogikken i Python er eksplisitt, og Spark følger RDD-avhengighetene:

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 sales_df har kolonner sku (streng) og 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)]

Her løser begge eksemplene det samme forretningsproblemet:

Filtrering av ugyldige salgsoppføringer og aggregering av gyldige mengder etter SKU.

Og i begge tilfeller collect() utløser Spark-kjøringen og returnerer alle resultatene til driveren.

Den viktigste forskjellen her ligger i hvordan resultatet representeres:

• RDD-API: Returnerer Python-elementer, for eksempel (sku, total_qty) tupler.
• DataFrame-API: Returnerer strukturerte Row objekter med navngitte felt, for eksempel sku og total_qty.

Her er en samlet sammenligning av RDD- og DataFrame-API-ene:

Tabell 2. RDD API vs. DataFrame API – sammenligning av funksjoner & abstraksjon: (Laget av Kuriko IWAI)

4. Avslutning

Målet med Spark-justering er ikke å maksimere antallet partisjoner eller CPU-bruken hver for seg.

Målet er å redusere den samlede kjørekostnaden og forsinkelsen uten å skape ustabilitet.

I denne artikkelen lærer vi om de grunnleggende sammenhengene i systemet:

• Ressurser: Arbeidernoder er vert for eksekutorene; eksekutorene kjører oppgaver; konfigurert CPU og minne begrenser samtidigheten.
• RDD-programmering: Transformasjoner produserer uforanderlige RDD-definisjoner og avstamningsinformasjon; handlinger utløser beregning.
• DataFrame-programmering: Transformasjoner produserer logiske spørringsoperasjoner; optimaliseringen velger kjørbare fysiske operatorer; handlinger utløser beregning.
• Kjøring: Spark organiserer arbeidet i jobber, faser og oppgaver, og oppgavene behandler partisjoner på eksekutorene. Shuffler og andre avhengigheter skaper koordineringsgrenser mellom fasene.

Å forstå hvordan Spark kjører distribuerte arbeidsbelastninger, er bare grunnlaget.

Du kan lære om en beregningsflyt i sju trinn ved hjelp av RDD API.