Hvad sker der egentlig, når du kører et Spark-job?

Indledning

Når datasæt vokser, kan det blive vanskeligt at behandle dem effektivt på én enkelt maskine på grund af begrænsninger i CPU-kapacitet, hukommelse og lagerets dataoverførselshastighed.

Apache Spark håndterer disse udfordringer ved at levere en distribueret databehandlingsmotor, der kan opdele arbejdsbelastninger i partitioner og udføre opgaver parallelt på tværs af CPU-kerner og flere maskiner.

Spark handler dog ikke blot om at tilføje flere maskiner.

Det er afgørende for at kunne designe effektive og pålidelige datapipelines at forstå, hvordan komponenterne arbejder sammen – drivere, eksekveringsprocesser, RDD’er, DataFrames, partitioner, jobs, stadier og opgaver.

I denne artikel gennemgår jeg Sparks arkitektur, udførelsesmekanik og centrale overvejelser om ydeevne med et praktisk eksempel på sammenlægning af e-handelssalg.

1. Hvorfor Apache Spark – parallel databehandling

Apache Spark er en distribueret beregningsmotor, der er udviklet til databehandling og data engineering i stor skala.

Spark behandler big data ved hjælp af parallel databehandling:

Figur A. Apache Sparks distribuerede klyngearkitektur på højt niveau (klynge-, driver- og workernoder) med sammenligning af udførelse på én node (oprettet af Kuriko IWAI)

Sammenlignet med behandling på én node (til højre, Figur A) opdeler Spark arbejdet i opgaver, der udføres på tværs af flere noder og – når ressourcerne er tilgængelige – på tværs af maskiner.

Her betyder distribueret ikke, at flere maskiner er påkrævet. Spark kan køre i lokal tilstand på én enkelt maskine. Selv dér kan data opdeles i partitioner og behandles parallelt.

2. Apache Sparks grundlæggende arkitektur

Spark består af flere centrale komponenter:

• Spark-driver (driver): Den centrale koordineringsproces i en Spark-applikation.
• Klyngeadministrator: En ekstern tjeneste, der har ansvaret for at allokere hardwareressourcer som CPU-kerner og hukommelse.
• Workernoder (worker): En fysisk eller virtuel maskine eller en beregningsnode i klyngen.
• Eksekveringsprocesser: Distribuerede arbejdsprocesser, der startes på workernoder.
• Java Virtual Machine (JVM): Det runtime-miljø, der kræves for at udføre Sparks bytekode (da Spark er skrevet i Scala/Java).

Først koordinerer driveren en Spark-applikation og omdanner den til en rettet acyklisk graf (DAG) af fysiske udførelsesopgaver, som den planlægger på tværs af eksekveringsprocesserne.

Derefter allokerer klyngeadministratoren ressourcer i henhold til implementeringens ressourcestyringssystem.

Workernoderne stiller beregningsressourcer til rådighed for eksekveringsprocesserne, som udfører opgaver og opbevarer data, mens opgaverne udføres.

Nogle administrerede miljøer, såsom Databricks, bruger én eksekveringsproces pr. worker, men det er ikke en universel Spark-regel. Afhængigt af klyngekonfigurationen kan en worker være vært for flere eksekveringsprocesser.

2-1. Anvendelse inden for e-handel – inputdatastruktur

Lad os forestille os en mindre e-handelsvirksomhed.

Den har 10.000 salgsoptegnelser, hver med en lagerføringsenhed (SKU) og en solgt mængde. Antag, at der er tre mulige SKU’er: A, B, og C.

Tabel 1. Eksempel på fordeling af e-handelssalgsoptegnelser og forventede sammenlægninger.

Målet er at beregne det samlede antal gyldigt solgte enheder pr. SKU:

1. Filtrér optegnelser fra, hvis qty <= 0, og derefter
2. Sammenlæg observationer efter SKU.

I det fulde datasæt i dette eksempel antages det, at 9.800 af de 10.000 optegnelser er tilbage efter filtreringen. De endelige totaler, som forventes fra Spark, er:

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

2-2. Spark-klynge

Jeg definerer en forenklet klynge til vores eksempel.

Workernoder (workers)

Jeg indstiller 3 workernoder, som hver har en instanstype med 8 vCPU’er og 32 GiB RAM:

Figur B. Ressourceallokering på tværs af workernoder (oprettet af Kuriko IWAI)

I alt er der 24 vCPU’er til rådighed i systemet.

Spark-executorer og tildeling af CPU og hukommelse

Lad os sige, at jeg reserverer:

• 7 executorer.
• 2 vCPU-kerner til hver executor.
• 3 GiB JVM-heap-hukommelse til hver executor.

Spark-applikationen anmoder klyngeadministratoren om at tildele ressourcer til disse 7 executorer.

Derefter beslutter klyngeadministratoren, hvordan CPU’er og hukommelse skal fordeles, på grundlag af hver workers ressourceforbrug.

For eksempel beslutter den at oprette 14 opgavepladser – 2 vCPU-kerner pr. executor x 7 executorer:

Figur C-1. vCPU-kerner, hukommelse og fordeling af executorer på tværs af workers med skæv ressourcefordeling på Worker A

Bemærk, at de 3 GiB hukommelse, der er tildelt en executor, ikke er dens samlede hukommelse, da der også skal reserveres hukommelse til overhead for workers og andre processer.

Hvis vi antager, at en enkelt opgave kræver 1 vCPU (spark.task.cpus = 1), tildeler hver worker sine vCPU’er:

Figur C-2. Parallel tildeling af vCPU’er på tværs af workers

Dette viser to vigtige punkter:

1. Hver worker har sin egen CPU-udnyttelsesgrad (i Figur C-2 er udnyttelsen på Worker A 75 %, mens Worker B og C fortsat ligger på 50 %).
2. Selvom systemet har 24 vCPU-kerner (8 vCPU’er pr. worker x 3 workers), kan det kun bruge 14 vCPU-kerner; de resterende 10 er uudnyttede til disse opgaver.

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

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

JVM er et runtime-miljø, der kræves for at afvikle Sparks bytekode.

Da Spark er skrevet i Scala/Java:

• Når der bruges Scala eller Java i applikationen: Koden afvikles direkte i JVM.
• Når der bruges PySpark (Python): En Python-proces afvikler koden, men kommunikerer via en bro som Py4J med en driver baseret på JVM. Spark-executorerne på workernoderne afvikles stadig i en JVM, som derefter styrer Python-processer, der afvikler Python-kode og brugerdefinerede Python-funktioner (Python UDF’er).

I begge tilfælde styrer JVM afviklingstråde, hukommelsestildeling og garbage collection for både driveren og executorerne.

Driver- og Executor-JVM’er i en PySpark-applikation

I en PySpark-applikation kommunikerer Python-driveren med Spark-driveren, der kører i JVM;

• Python-driver: Her skrives og afvikles PySpark-koden på applikationsniveau.
• Driver-JVM: Koordinerer afviklingen af Spark og planlægningen.
• Executor-JVM: Afvikler Spark-opgaver og administrerer afviklingsressourcer. Python-processer indgår, når opgaver kræver afvikling af Python-kode.
• Klient-JVM (klientproces eller indsendelsesmaskine): JVM, der kører på den lokale maskine eller en edge-/gatewaynode.

Som vist i figuren opretter klient-JVM’en først forbindelse til driver-JVM’en, som derefter koordinerer med executor-JVM’erne:

Figur D. Overordnet sammenhæng mellem klient-, driver- og executor-JVM’er

Herfra er Spark-driveren placeret i driver-JVM’en, hvor den administrerer applikationen, mens de enkelte executorer er placeret i executor-JVM’erne og står for at afvikle opgaverne.

Hver driver-JVM har en JVM-heap, der effektivt håndterer databehandling, caching og brugerobjekter.

2-4. Dybdegående gennemgang af Resilient Distributed Datasets (RDD’er)

Et Resilient Distributed Dataset (RDD) er en uforanderlig, partitioneret samling af elementer, som Spark kan behandle parallelt.

RDD’er registrerer afhængigheder mellem transformationer, så Spark kan genberegne mistede partitioner i mange fejlsituationer.

Navnet afspejler tre egenskaber:

• Robust: Spark registrerer afstamningen – altså hvilket RDD der afhænger af hvilket tidligere RDD – og kan ofte genskabe manglende partitioner i stedet for at vedligeholde replikaer af hvert mellemliggende resultat.
• Distribueret: Datasættet repræsenteres som partitioner, som Spark kan behandle på forskellige executorer. Det er ikke et krav at bruge flere fysiske maskiner; RDD’er understøttes også i lokal tilstand.
• Datasæt: Et RDD kan indeholde nøgle-værdi-par, heltal, strenge, tupler og andre understøttede elementer – ikke kun relationelle rækker. RDD-API’erne omfatter operationer som map, filter, reduceByKey, join, og aggregateByKey.

En RDD er en logisk samling og beregningsabstraktion, ikke en erklæring om, at bestemte partitioner permanent er tildelt bestemte arbejdere.

Partitioner

En partition er en logisk delmængde af data og svarer normalt til én opgave i et trin.

Lad os sige, at vi opretter 100 partitioner (disse 100 indledende partitioner er valgt specifikt til undervisningsformålet).

Med 10.000 kildeelementer giver det cirka 100 elementer pr. partition, hvis inputtet er fordelt nogenlunde jævnt:

Figur E. Parallel fordeling af partitioner på tværs af 100 indledende dataopdelinger

3. RDD-API vs. DataFrame-API: Catalysts interne optimering

Det oprindelige RDD-eksempel gør det lettere at forstå lineage, partitioner og shuffle.

I typiske strukturerede ETL-processer er DataFrame-API’et dog det foretrukne udgangspunkt, fordi det giver adgang til kolonner, skemainformation, SQL-lignende udtryk og operationer, som optimeringsværktøjet kan se.

Diagrammet nedenfor illustrerer sammenhængen:

Figur F. Sammenhængen mellem Sparks grafkoncepter: RDD-lineage, DataFrame og Sparks udførelses-DAG

Ligesom en RDD har RDD-lineage, der beskriver transformationerne, repræsenterer en DataFrame-forespørgselsplan den strukturerede forespørgsel.

Spark opbygger udførelsen ud fra afhængigheder og fysiske operatorer.

DataFrame-udførelse kan internt benytte RDD-baserede mekanismer, men det betyder ikke, at brugerne selv skal opbygge RDD-lineage, når de skriver DataFrame-kode.

3-1. DataFrame-forespørgselsplan

En DataFrame-forespørgselsplan er den sekvens af logiske og fysiske trin, som en udførelsesmotor genererer (den lilla boks i Figur L).

En DataFrame-forespørgsel gennemløber flere repræsentationer:

1. Logisk plan: En beskrivelse af de ønskede handlinger (scanning, filtrering, gruppering, aggregering).
2. Analyseret og optimeret logisk plan: Spark afklarer referencer og anvender relevante logiske optimeringer, såsom kolonnebeskæring eller predicate pushdown.
3. Fysisk plan: Spark vælger eksekverbare operatorer, f.eks. scanninger, delvise/endelige aggregeringer, udvekslinger, joins og sorteringer.
4. Udførelse: Den fysiske plan køres ved hjælp af Sparks udførelsesmekanismer og involverer i sidste ende opgaver på executor-processer.

I vores anvendelsesscenarie kan den fysiske plan omfatte følgende trin:

• Delvis hashaggregering,
• udveksling efter SKU og
• endelig hashaggregering.

3-2. Sammenligning af handlinger: RDD-API vs. DataFrame-API

Lad os sammenligne RDD-API’et og DataFrame-API’et med aggregering efter SKU som eksempel.

RDD-API

Elementerne er (sku, qty) par. Vores aggregeringslogik i Python er eksplicit, og Spark følger RDD-afhængighederne:

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

Antag, at sales_df har kolonnerne sku (tekst) 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 eksempler det samme forretningsproblem:

Filtrering af ugyldige salgsregistreringer og aggregering af gyldige mængder efter SKU.

Og i begge tilfælde collect() udløser Spark-udførelsen og returnerer alle de resulterende poster til driveren.

Den væsentligste forskel er her, hvordan resultatet repræsenteres:

• RDD-API: Returnerer Python-elementer, f.eks. (sku, total_qty) tupler.
• DataFrame-API: Returnerer strukturerede Row objekter med navngivne felter, f.eks. sku og total_qty.

Her er en samlet sammenligning af RDD- og DataFrame-API’erne:

Tabel 2. RDD API vs. DataFrame API – sammenligning af funktioner & abstraktionsniveau:(Udarbejdet af Kuriko IWAI)

4. Afslutning

Målet med tuning af Spark er ikke isoleret set at maksimere antallet af partitioner eller CPU-forbruget.

Målet er at reducere eksekveringsomkostningerne og latenstiden fra start til slut uden at skabe ustabilitet.

I denne artikel gennemgår vi de grundlæggende sammenhænge i systemet:

• Ressourcer: Workers er værter for executors; executors kører opgaver; den konfigurerede CPU og hukommelse begrænser samtidigheden.
• RDD-programmering: Transformationer producerer uforanderlige RDD-definitioner og lineage; handlinger udløser beregning.
• DataFrame-programmering: Transformationer producerer logiske forespørgselsoperationer; optimeringen vælger eksekverbare fysiske operatorer; handlinger udløser beregning.
• Eksekvering: Spark organiserer arbejdet i jobs, stadier og opgaver, og opgaver behandler partitioner på executors. Shuffles og andre afhængigheder skaber koordinationsgrænser mellem stadierne.

At forstå, hvordan Spark eksekverer distribuerede arbejdsbelastninger, er kun grundlaget.

Du kan lære et beregningsforløb i syv trin med RDD API.