Mitä Spark-työn suorittamisen aikana oikeastaan tapahtuu?

Johdanto

Kun tietoaineistot kasvavat, niiden tehokas käsittely yhdellä koneella voi olla haastavaa suorittimen kapasiteetin, muistin ja tallennusnopeuden rajoitusten vuoksi.

Apache Spark vastaa näihin haasteisiin tarjoamalla hajautetun tiedonkäsittelymoottorin, joka jakaa työkuormat osioihin ja suorittaa tehtävät rinnakkain suorittimen ytimillä ja useilla koneilla.

Sparkissa ei kuitenkaan ole kyse pelkästään koneiden lisäämisestä.

Jotta tehokkaita ja luotettavia tietoputkia voidaan suunnitella, on tärkeää ymmärtää, miten sen osat toimivat yhdessä: ohjaimet, suorittajat, RDD:t, DataFrame-kehykset, osiot, työt, vaiheet ja tehtävät.

Tässä artikkelissa käyn läpi Sparkin arkkitehtuuria, suoritusmekanismeja ja keskeisiä suorituskykyyn vaikuttavia tekijöitä käytännön esimerkin avulla, jossa verkkokaupan myyntitietoja koostetaan.

1. Miksi Apache Spark – rinnakkaista tiedonkäsittelyä

Apache Spark on hajautetun laskennan moottori, joka on suunniteltu suurten tietomäärien käsittelyyn ja tietotekniikan kehitystyöhön.

Spark käsittelee suurdataa hyödyntämällä rinnakkaista tiedonkäsittelyä:

Kuva A. Apache Sparkin hajautetun klusterin korkean tason arkkitehtuuri (klusteri, ohjain ja työntekijäsolmut), jossa verrataan suoritusta yhdellä solmulla (luonut Kuriko IWAI)

Yhden solmun käsittelyyn verrattuna (oikealla, kuvassa A) Spark jakaa työn tehtäviksi useiden solmujen kesken ja suorittaa ne käytettävissä olevilla resursseilla eri koneilla.

Hajautettu ei tässä tarkoita, että useat koneet olisivat välttämättömiä. Sparkia voi käyttää paikallisessa tilassa yhdellä koneella. Silloinkin tiedot voidaan jakaa osioihin ja käsitellä rinnakkain.

2. Apache Sparkin keskeinen arkkitehtuuri

Spark koostuu useista keskeisistä osista:

• Spark-ohjain (ohjain): Spark-sovelluksen keskeinen koordinointiprosessi.
• Klusterinhallinta: Ulkoinen palvelu, joka vastaa laitteistoresurssien, kuten suoritinytimien ja muistin, varaamisesta.
• Työntekijäsolmut (työntekijä): Klusteriin kuuluva fyysinen tai virtuaalinen kone tai laskentasolmu.
• Suorittajat: Työntekijäsolmuissa käynnistettävät hajautetut työprosessit.
• Java Virtual Machine (JVM): Suoritusympäristö, jota tarvitaan Sparkin tavukoodin suorittamiseen (koska Spark on kirjoitettu Scalalla ja Javalla).

Ensin ohjain koordinoi Spark-sovellusta ja muuntaa sen fyysisesti suoritettavien tehtävien suunnatuksi syklittömäksi graafiksi (DAG) sekä aikatauluttaa tehtävät suorittajille.

Tämän jälkeen klusterinhallinta varaa resursseja käyttöönoton resurssienhallintajärjestelmän puitteissa.

Työntekijäsolmut tarjoavat suorittajille laskentaresursseja. Suorittajat suorittavat tehtäviä ja säilyttävät tietoja tehtävien suorittamisen aikana.

Joissakin hallituissa ympäristöissä, kuten Databricksissa, kutakin työntekijää kohden on yksi suorittaja, mutta tämä ei ole kaikissa Spark-ympäristöissä noudatettava sääntö. Klusterin kokoonpanosta riippuen yhdessä työntekijässä voi olla useita suorittajia.

2–1. Verkkokaupan käyttötapaus – syötetietojen rakenne

Kuvitellaan pieni verkkokauppayritys.

Sillä on 10 000 myyntitietuetta, joista jokaisessa on varastonimike (SKU) ja myyty määrä. Oletetaan, että mahdollisia SKU-tunnuksia on kolme: A, B, ja C.

Taulukko 1. Esimerkki verkkokaupan myyntitietueiden jakaumasta ja odotetuista koostetuista tuloksista.

Tavoitteena on laskea kunkin SKU:n perusteella myytyjen kelvollisten yksiköiden kokonaismäärä:

1. Suodata pois tietueet, joiden qty <= 0, ja sitten
2. Koosta otokset SKU:n mukaan.

Oletetaan havainnollistavassa koko tietoaineistossa, että suodatuksen jälkeen jäljelle jää 9 800 tietuetta 10 000 tietueesta. Sparkin odotetut lopulliset summat ovat:

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

2–2. Spark-klusteri

Määrittelen käyttötapausta varten yksinkertaistetun klusterin.

Työntekijäsolmut (työntekijät)

Asetan klusteriin 3 työntekijäsolmua, joiden kunkin instanssityyppi on 8 vCPU:ta ja joissa on 32 GiB RAM-muistia:

Kuva B. Resurssien jako työntekijäsolmujen kesken (luonut Kuriko IWAI)

Järjestelmässä on yhteensä käytettävissä 24 vCPU:ta.

Spark-suorittimet sekä suorittimien ja muistin varaus

Oletetaan, että varaan:

• 7 suoritinta.
• 2 vCPU-ydintä kutakin suoritinta kohti.
• 3 GiB JVM-keomuistia kutakin suoritinta kohti.

Spark-sovellus pyytää klusterinhallintaa varaamaan resursseja näille 7 suorittimelle.

Tämän jälkeen klusterinhallinta päättää suoritinten ja muistin varauksista kunkin työntekijän resurssien käytön perusteella.

Se päättää esimerkiksi luoda 14 tehtäväpaikkaa – 2 vCPU-ydintä suoritinta kohti × 7 suoritinta:

Kuva C-1. vCPU-ytimien, muistin ja suoritinten varaus työntekijöiden kesken, kun työntekijän A resurssit ovat epätasapainossa

Huomaa, että suorittimelle varattu 3 GiB:n muisti ei ole sen kokonaismuisti, sillä työntekijöille ja muille suorituksille on varattava lisäksi muistin lisätilaa.

Oletetaan, että yksi tehtävä tarvitsee 1 vCPU:n (spark.task.cpus = 1), jolloin kukin työntekijä varaa vCPU-ytimensä:

Kuva C-2. vCPU-ytimien rinnakkainen varaus työntekijöiden kesken

Tämä havainnollistaa kahta keskeistä asiaa:

1. Kullakin työntekijällä on yksilöllinen suorittimen käyttöaste (Kuvassa C-2 työntekijän A käyttöaste on 75 %, kun taas työntekijöiden B ja C käyttöaste pysyy 50 prosentissa).
2. Vaikka järjestelmässä on 24 vCPU-ydintä (8 vCPU:ta työntekijää kohti × 3 työntekijää), se voi käyttää niistä vain 14:ää; loput 10 jäävät näissä tehtävissä käyttämättä.

2-3. Sparkin taustalla toimiva Java-virtuaalikone (JVM)

Apache Spark perustuu Java-virtuaalikoneeseen (JVM).

JVM on Spark-tavukoodin suorittamiseen tarvittava ajoitusympäristö.

Koska Spark on kirjoitettu Scalalla ja Javalla:

• Kun sovelluksessa käytetään Scalaa tai Javaa: koodi suoritetaan natiivisti JVM:n sisällä.
• Kun käytetään PySparkia (Pythonia): Python-prosessi suorittaa koodin, mutta viestii JVM-pohjaisen ohjaimen kanssa esimerkiksi Py4J:n välityksellä. Työntekijäsolmujen Spark-suorittimet toimivat edelleen JVM:n sisällä, joka hallinnoi Python-työntekijöitä Python-koodin ja Pythonin käyttäjän määrittämien funktioiden (Python UDF:t) suorittamista varten.

Kummassakin tapauksessa JVM hallinnoi sekä ohjaimen että suoritinten suoritussäikeitä, muistin varausta ja roskienkeruuta.

PySpark-sovelluksen ohjaimen ja suoritinten JVM:t

PySpark-sovelluksessa Python-ohjain viestii JVM:ssä toimivan Spark-ohjaimen kanssa;

• Python-ohjain: paikka, jossa sovellustason PySpark-koodi kirjoitetaan ja suoritetaan.
• Ohjaimen JVM: koordinoi Spark-suoritusta ja ajoitusta.
• Suorittimen JVM: suorittaa Spark-tehtäviä ja hallinnoi suorituksen resursseja. Kun tehtävät edellyttävät Python-koodin suorittamista, mukana on Python-työntekijäprosesseja.
• Asiakas-JVM (asiakasprosessi tai lähetyskone): paikallisessa koneessa tai reuna-/yhdyskäytäväsolmussa toimiva JVM.

Kuvan mukaisesti asiakas-JVM muodostaa ensin yhteyden ohjaimen JVM:ään, joka puolestaan koordinoi toimintaa suoritinten JVM:ien kanssa:

Kuva D. Asiakkaan, ohjaimen ja suoritinten JVM:ien yleinen suhde

Tästä eteenpäin ohjaimen JVM isännöi Spark-ohjainta, joka hallinnoi sovellusta, kun taas suoritinten JVM:t isännöivät yksittäisiä suorittimia, jotka vastaavat näiden tehtävien suorittamisesta.

Kunkin ohjaimen JVM:ssä on JVM-keomuisti, jonka avulla tietojenkäsittely, välimuistitus ja käyttäjäoliot voidaan hoitaa tehokkaasti.

2-4. Resilient Distributed Dataset (RDD) -tietorakenteiden syvempi tarkastelu

Resilient Distributed Dataset (RDD) on muuttumaton, osioitu alkioiden kokoelma, jota Spark voi käsitellä rinnakkain.

RDD:t seuraavat muunnosten välisiä riippuvuuksia, minkä ansiosta Spark voi laskea kadonneet osiot uudelleen monissa virhetilanteissa.

Nimi kuvaa kolmea ominaisuutta:

• Vikasietoinen: Spark tallentaa riippuvuushistorian – sen, mistä aiemmasta RDD:stä kukin RDD riippuu – ja pystyy usein palauttamaan puuttuvat osiot sen sijaan, että se ylläpitäisi kopioita jokaisesta välituloksesta.
• Hajautettu: Tietojoukko koostuu osioista, joita Spark voi käsitellä eri suorittimilla. Useita fyysisiä koneita ei tarvita; myös paikallinen tila tukee RDD:itä.
• Tietojoukko: RDD voi sisältää avain-arvo-pareja, kokonaislukuja, merkkijonoja, monikkoja ja muita tuettuja alkioita – ei ainoastaan relaatiotaulukoiden rivejä. RDD-rajapintoihin kuuluu esimerkiksi toimintoja, kuten map, filter, reduceByKey, join, ja aggregateByKey.

RDD on looginen kokoelman ja laskennan abstraktio, ei määritelmä siitä, että tietyt osiot olisi pysyvästi osoitettu tietyille työntekijöille.

Osiot

Osio on datan looginen osa, joka vastaa yleensä yhtä tehtävää vaiheen sisällä.

Oletetaan, että luomme 100 osiota (nämä 100 alkuosiota ovat tässä opetuksellisista syistä tehty tietoinen valinta).

Kun lähde-elementtejä on 10 000, kuhunkin osioon tulee noin 100 elementtiä, jos syöte jakautuu osioihin suunnilleen tasaisesti:

Kuva E. Osioiden rinnakkainen jakaminen 100 alkuperäiseen dataosioon

3. RDD-rajapinta vs. DataFrame-rajapinta: Catalystin sisäinen optimointi

Alkuperäinen RDD-esimerkki auttaa ymmärtämään tietovirtaa, osioita ja sekoitusta.

Tyypillisessä rakenteisessa ETL-käsittelyssä DataFrame-rajapinta on kuitenkin suositeltava lähtökohta, sillä se tarjoaa sarakkeet, skeematiedot, SQL-tyyppiset lausekkeet ja optimoijan hyödynnettävissä olevat operaatiot.

Alla oleva kaavio havainnollistaa niiden suhdetta:

Kuva F. Spark-kaavion käsitteiden väliset suhteet: RDD:n tietovirta, DataFrame ja Sparkin suoritus-DAG

Aivan kuten RDD:n tietovirta määrittää muunnokset, DataFrame-kyselysuunnitelma kuvaa rakenteisen kyselyn.

Spark muodostaa suoritettavat työt riippuvuuksien ja fyysisten operaattorien perusteella.

DataFramen suoritus voi käyttää sisäisesti RDD-pohjaisia mekanismeja, mutta se ei tarkoita, että käyttäjien täytyisi rakentaa RDD:n tietovirta käsin DataFrame-koodia kirjoittaessaan.

3-1. DataFrame-kyselysuunnitelma

DataFrame-kyselysuunnitelma on suoritusmoottorin muodostama loogisten ja fyysisten vaiheiden sarja (kuvan L violetti laatikko).

DataFrame-kysely käy läpi useita esitysmuotoja:

1. Looginen suunnitelma: Kuvaus pyydetyistä operaatioista (luku, suodatus, ryhmittely, koonti).
2. Analysoitu ja optimoitu looginen suunnitelma: Spark ratkaisee viittaukset ja soveltaa mahdollisia loogisia optimointeja, kuten sarakkeiden karsimista tai predikaattien siirtoa.
3. Fyysinen suunnitelma: Spark valitsee suoritettavat operaattorit, kuten luvut, osittaiset ja lopulliset koonnit, vaihdot, liitokset ja lajittelut.
4. Suoritus: Fyysinen suunnitelma suoritetaan Sparkin suoritusmekanismeilla, ja siihen kuuluu lopulta suorittajien tehtäviä.

Käyttötapauksessamme fyysinen suunnitelma voi sisältää seuraavat vaiheet:

• Osittainen hajautustaulukkokoonti,
• vaihto SKU:n mukaan ja
• lopullinen hajautustaulukkokoonti.

3-2. Toimintojen vertailu: RDD-rajapinta vs. DataFrame-rajapinta

Verrataan RDD- ja DataFrame-rajapintoja käyttämällä SKU-koontia esimerkkinä.

RDD-rajapinta

Elementit ovat (sku, qty) pareja. Python-koontilogiikkamme on eksplisiittinen, ja Spark seuraa RDD-riippuvuuksia:

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-rajapinta

Oletetaan, että sales_df sisältää sarakkeet sku (merkkijono) ja qty (numeerinen):

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

Molemmilla esimerkeillä ratkaistaan sama liiketoimintaongelma:

Virheellisten myyntitietueiden suodattaminen ja kelvollisten määrien kokoaminen SKU:n mukaan.

Molemmissa tapauksissa collect() käynnistää Sparkin suorituksen ja palauttaa kaikki tuloksena saadut tietueet ajurille.

Keskeinen ero on tässä tuloksen esitysmuodossa:

• RDD-rajapinta: Palauttaa Python-elementtejä, kuten (sku, total_qty) monikoita.
• DataFrame-rajapinta: Palauttaa rakenteisia Row nimettykenttäisiä olioita, kuten sku ja total_qty.

Tässä on RDD- ja DataFrame-rajapintojen yleisvertailu:

Taulukko 2. RDD API:n ja DataFrame API:n ominaisuuksien ja abstraktiotason vertailu (laatija: Kuriko IWAI)

4. Yhteenveto

Sparkin virityksen tavoitteena ei ole maksimoida osioiden määrää tai suorittimen käyttöastetta toisistaan riippumatta.

Tavoitteena on pienentää suorituksen kokonaiskustannuksia ja viivettä aiheuttamatta epävakautta.

Tässä artikkelissa perehdymme järjestelmän keskeisiin yhteyksiin:

• Resurssit: Työntekijäsolmut isännöivät suorittajia; suorittajat suorittavat tehtäviä; määritetyt suoritin- ja muistiresurssit rajoittavat samanaikaisuutta.
• RDD-ohjelmointi: Muunnokset tuottavat muuttumattomia RDD-määrityksiä ja riippuvuusketjuja; toiminnot käynnistävät laskennan.
• DataFrame-ohjelmointi: Muunnokset tuottavat loogisia kyselyoperaatioita; optimointi valitsee suoritettavat fyysiset operaattorit; toiminnot käynnistävät laskennan.
• Suoritus: Spark järjestää työn töiksi, vaiheiksi ja tehtäviksi, ja tehtävät käsittelevät osioita suorittajilla. Shufflet ja muut riippuvuudet muodostavat vaiheiden välisiä koordinointirajoja.

Sen ymmärtäminen, miten Spark suorittaa hajautettuja työkuormia, on vasta perusta.

Voit tutustua RDD APIa käyttävään seitsemänvaiheiseen laskentakulkuun.