Apache Spark: Különbségek az adatkeretek, az adatkészletek és az RDD között

1. Áttekintés

Az Apache Spark egy gyors, elosztott adatfeldolgozó rendszer. A memóriában történő adatfeldolgozást végzi, a memória gyorsítótárát és az optimalizált végrehajtást használja, amely gyors teljesítményt eredményez. Magas szintű API-kat biztosít olyan népszerű programnyelvekhez, mint a Scala, Python, Java és R.

Ebben a gyors oktatóanyagban a Spark három alapkoncepcióján megyünk keresztül: adatkeretek, adatkészletek és RDD-k.

2. DataFrame

A Spark SQL a Spark 1.3 óta DataFrame nevű táblázatos adatkivonatot vezetett be. Azóta a Spark egyik legfontosabb jellemzőjévé vált. Ez az API akkor hasznos, ha strukturált és félig strukturált, elosztott adatokat akarunk kezelni.

A 3. szakaszban tárgyaljuk a Rugalmas elosztott adatkészleteket (RDD). A DataFrames hatékonyabban tárolja az adatokat, mint az RDD-k, mert ezek az RDD-k megváltoztathatatlan, memóriában lévő, rugalmas, elosztott és párhuzamos képességeit használják, de sémát is alkalmaznak az adatokra. A DataFrames az SQL-kódot optimalizált alacsony szintű RDD-műveletekké is fordítja.

Három módon hozhatunk létre DataFrame-eket:

  • Meglévő RDD-k konvertálása
  • SQL lekérdezések futtatása
  • Külső adatok betöltése

Spark csapat bemutatta SparkSession a 2.0 verzióban egyesíti az összes különböző kontextust, biztosítva, hogy a fejlesztőknek ne kelljen aggódniuk különböző kontextusok létrehozása miatt:

SparkSession munkamenet = SparkSession.builder () .appName ("TouristDataFrameExample") .master ("local [*]") .getOrCreate (); DataFrameReader dataFrameReader = session.read ();

Elemezzük a Tourist.csv fájl:

Adatkészlet adat = dataFrameReader.option ("header", "true") .csv ("data / Tourist.csv");

Amióta a Spark 2.0 DataFrame a Adatkészlet típusú Sor, így a DataFrame-et használhatjuk álnévként a Adatkészlet.

Kiválaszthatunk konkrét oszlopokat, amelyek érdekelnek. Szűrhetünk és csoportosíthatunk is egy adott oszlop szerint:

data.select (col ("ország"), col ("év"), col ("érték")) .show (); data.filter (col ("ország"). equTo ("Mexikó")) .show (); data.groupBy (col ("ország")) .szám () .show ();

3. Adatkészletek

Az adatkészlet erősen tipizált, strukturált adatok összessége. Biztosítják az ismert objektum-orientált programozási stílust, valamint a típusbiztonság előnyeit, mivel az adatkészletek fordítási időben ellenőrizhetik a szintaxist és a hibákat.

Adatkészlet a DataFrame kiterjesztése, így a DataFrame-et egy adatkészlet be nem írt nézetének tekinthetjük.

A Spark csapata kiadta a Adatkészlet API a Spark 1.6-ban és mint említettük: „a Spark Datasets célja egy olyan API biztosítása, amely lehetővé teszi a felhasználók számára, hogy egyszerűen kifejezzék az átalakításokat az objektumtartományokban, ugyanakkor a Spark SQL végrehajtó motor teljesítményének és robusztusságának előnyeit is biztosítják”.

Először létre kell hoznunk egy típusú osztályt TouristData:

public class TouristData {privát String régió; privát String ország; magán String év; privát húr sorozat; magán Dupla érték; saját karakterlánc lábjegyzetek; privát String forrás; // ... getterek és beállítók}

Az egyes rekordjainknak a megadott típushoz való hozzárendeléséhez kódolót kell használnunk. A kódolók fordítanak a Java objektumok és a Spark belső bináris formátuma között:

// SparkSession inicializálása és adatbetöltése Adatkészlet responseWithSelectedColumns = data.select (col ("régió"), col ("ország"), col ("év"), col ("sorozat"), col ("érték") cast. ("kettős"), col ("lábjegyzetek"), col ("forrás")); Dataset typedDataset = responseWithSelectedColumns .as (Encoders.bean (TouristData.class));

A DataFrame-hez hasonlóan szűrhetünk és csoportosíthatunk egyes oszlopok szerint:

typedDataset.filter ((FilterFunction) rekord -> record.getCountry () .equals ("Norvégia")) .show (); typedDataset.groupBy (typedDataset.col ("ország")) .count () .show ();

Olyan műveleteket is végezhetünk, mint például egy adott tartománynak megfelelő oszlop szerinti szűrés vagy egy adott oszlop összegének kiszámítása, hogy megkapjuk annak teljes értékét:

typedDataset.filter ([FilterFunction] record -> record.getYear ()! = null && (Long.valueOf (record.getYear ())> 2010 && Long.valueOf (record.getYear ()) record.getValue ()! = null && record.getSeries () .contains ("ráfordítás")) .groupBy ("ország") .agg (összeg ("érték")) .show ();

4. RDD-k

A Rugalmas elosztott adatkészlet vagy az RDD a Spark elsődleges programozási absztrakciója. Olyan elemek gyűjteményét jelenti, amelyek: változhatatlan, rugalmas és elosztott.

Az RDD egy nagy adatkészletet foglal magában, a Spark automatikusan elosztja az RDD-kben található adatokat a fürtünkön, és párhuzamosan végzi a velük végzett műveleteket..

RDD-ket csak stabil adattárban lévő adatok vagy más RDD-k műveletei révén hozhatunk létre.

A hibatűrés elengedhetetlen, ha nagy adathalmazokkal foglalkozunk, és az adatokat fürtgépeken osztják szét. Az RDD-k a Spark beépített hibahelyezési mechanikája miatt rugalmasak. A Spark arra a tényre támaszkodik, hogy az RDD-k megjegyzik létrehozásuk módját, hogy a partíció helyreállítása érdekében könnyen visszavezethessük a származást..

Kétféle műveletet tudunk elvégezni az RDD-ken: Átalakítások és cselekvések.

4.1. Átalakulások

Alkalmazhatunk transzformációkat egy RDD-re annak adatainak manipulálására. A manipuláció végrehajtása után egy teljesen új RDD-t kapunk, mivel az RDD-k megváltoztathatatlan objektumok.

Megvizsgáljuk a Map and Filter megvalósítását, amely a két leggyakoribb átalakítás.

Először létre kell hoznunk a JavaSparkContext és töltse be az adatokat RDD-ként a Tourist.csv fájl:

SparkConf conf = new SparkConf (). SetAppName ("uppercaseCountries") .setMaster ("local [*]"); JavaSparkContext sc = új JavaSparkContext (conf); JavaRDD turisták = sc.textFile ("data / Tourist.csv");

Ezután alkalmazzuk a térképfunkciót, hogy minden rekordból megkapjuk az ország nevét, és alakítsuk át a nevet nagybetűvé. Ezt az újonnan létrehozott adatkészletet szöveges fájlként menthetjük a lemezre:

JavaRDD upperCaseCountries = turisták.map (sor -> {String [] oszlopok = line.split (COMMA_DELIMITER); visszatérő oszlopok [1] .toUpperCase ();}). Külön (); upperCaseCountries.saveAsTextFile ("adatok / kimenet / nagybetűs.txt");

Ha csak egy adott országot akarunk kiválasztani, akkor alkalmazhatjuk a szűrő funkciót az eredeti turistáink RDD-jére:

JavaRDD turistákInMexico = turisták .filter (vonal -> line.split (COMMA_DELIMITER) [1] .egyenlő ("Mexikó")); turistákInMexico.saveAsTextFile ("adatok / output / touristInMexico.txt");

4.2. Műveletek

A műveletek visszaadják a végső értéket, vagy az eredményeket lemezre mentik, miután valamilyen számítást végeztek az adatokon.

A Sparkban a visszatérően használt műveletek közül kettő a Count és a Reduce.

Számoljuk meg az összes országot a CSV-fájlunkban:

// Spark Context inicializálása és adatterhelés JavaRDD országok = turisták.map (sor -> {String [] oszlopok = line.split (COMMA_DELIMITER); visszatérő oszlopok [1];}). Long numberOfCountries = országok.szám ();

Most országonként kiszámoljuk a teljes kiadást. Szűrnünk kell a kiadásokat tartalmazó leírásokat.

Ahelyett, hogy a JavaRDD, használjuk a JavaPairRDD. Az RDD pár egy olyan RDD típus, amely kulcs-érték párokat képes tárolni. Ellenőrizzük a következőt:

JavaRDD turistákKiadások = turisták .szűrő (sor -> sor.split (COMMA_DELIMITER) [3] .tartalmaz ("kiadások")); JavaPairRDD kiadásokPairRdd = turistákKiadások .mapToPair (sor -> {String [] oszlopok = line.split (COMMA_DELIMITER); adja vissza az új Tuple2-t (oszlopok [1], Double.valueOf (oszlopok [6]);}); Lista totalByCountry = kiadásPárRdd .reduceByKey ((x, y) -> x + y) .collect ();

5. Következtetés

Összefoglalva: ha tartományspecifikus API-ra van szükségünk, akkor DataFrame-eket vagy adatkészleteket kell használnunk, olyan magas szintű kifejezésekre van szükségünk, mint például összesítés, összeg vagy SQL-lekérdezések. Vagy amikor a típusbiztonságot akarjuk a fordítás idején.

Másrészt RDD-ket kell használnunk, ha az adatok strukturálatlanok, és nincs szükségünk egy adott séma megvalósítására, vagy amikor alacsony szintű átalakításokra és műveletekre van szükségünk.

Mint mindig, az összes kódminta elérhető a GitHubon.