Wat is Spark Shell-opdrachten?
Spark shell is een interface die wordt gebruikt om adhoc-vragen te schrijven om te werken en het gedrag van Apache Spark te begrijpen. Het wordt de cluster computing open-source engine genoemd die gegevens in het geheugen kan verwerken, zoals voor analyse, ETL, machine learning voor grote hoeveelheden gegevens. In dit onderwerp gaan we meer te weten komen over Spark Shell-opdrachten.
Er zijn verschillende soorten Spark-shell voor verschillende programmeertalen bijvoorbeeld:
- spark-shell is geschreven in Scala
- pyspark bevindt zich in Python en
- sparkR voor R-taal
Men kan zijn zelfstandige toepassing ontwikkelen met behulp van Spark. Het wordt veel gebruikt vanwege zijn supersnelle berekeningssnelheid. Dit komt omdat het MapReduce gebruikt om verschillende query's en transformaties te verwerken.
Om spark-shell-opdrachten uit te voeren, moeten Java en Scala al in het systeem zijn geïnstalleerd.
Soorten Spark Shell-opdrachten
De verschillende soorten Spark-shell-opdrachten zijn als volgt:
1. Om te controleren of de Spark is geïnstalleerd en om te weten wat de versie is, wordt het onderstaande commando gebruikt (Alle commando's hierna worden aangegeven beginnend met dit symbool "$")
$ spark-shell
De volgende uitgang wordt weergegeven als de vonk is geïnstalleerd:
$ spark-shell
SPARK_MAJOR_VERSION is ingesteld op 2 met Spark2
Het standaard logniveau instellen op "WARN".
Gebruik sc.setLogLevel (newLevel) om het logboekniveau aan te passen. Gebruik voor SparkR setLogLevel (newLevel).
Spark context Web UI beschikbaar op http://10.113.59.34:4040
Spark context beschikbaar als 'sc' (master = local (*), app id = local-1568732886588).
Spark-sessie beschikbaar als 'spark'.
Welkom bij
____ __
/ __ / __ ___ _____ / / __
_ \ \ / _ \ / _ `/ __ / '_ /
/ ___ / .__ / \ _, _ / _ / / _ / \ _ \ versie 2.2.0.2.6.3.0-235
/ _ /
Scala-versie 2.11.8 gebruiken (Java HotSpot (TM) 64-Bit Server VM, Java 1.8.0_112)
Typ expressies om ze te laten evalueren.
Type: help voor meer informatie.
scala>
2. De basisgegevensstructuur van Spark wordt een RDD (Resilient Distributed Datasets) genoemd die een onveranderlijke verzameling objecten bevat voor gedistribueerde gegevensverwerking van records. Alle datasets van RDD zijn logisch verdeeld over meerdere knooppunten van een cluster.
Een RDD kan alleen worden gemaakt door te lezen van een lokaal bestandssysteem of door een bestaande RDD te transformeren.
a) Om een nieuwe RDD te maken gebruiken we de volgende opdracht:
scala> val examplefile = sc.textFile("file.txt")
Hier wordt sc het object van SparkContext genoemd.
Output:
examplefile: org.apache.spark.rdd.RDD(String) = file.txt MapPartitionsRDD(3) at textFile at :24
b) Een RDD kan als volgt via Parallelized Collection worden gemaakt:
scala> val oddnum = Array(1, 3, 5, 7, 9)
Output:
oddnum: Array(Int) = Array(1, 3, 5, 7, 9)
scala> val value = sc.parallelize(oddnum)
Output:
value: org.apache.spark.rdd.RDD(Int) = ParallelCollectionRDD(4) at parallelize at :26
c) Maken van bestaande RDD's :
scala> val newRDD = oddnum.map(value => (value * 2))
Output:
newRDD: Array(Int) = Array(2, 6, 10, 14, 18)
3. Er zijn twee soorten Spark RDD-bewerkingen die kunnen worden uitgevoerd op de gemaakte datasets:
- acties
- transformaties
Acties: het wordt gebruikt om bepaalde vereiste bewerkingen op de bestaande gegevenssets uit te voeren. Hier volgen enkele van de opdrachten die kunnen worden gebruikt om de onderstaande acties op de gemaakte gegevenssets uit te voeren:
a) count () functie om het aantal elementen in RDD te tellen:
scala> value.count()
Output:
res3: Long = 5
b) collect () functie om alle elementen van de array weer te geven:
scala> value.collect()
Output:
res5: Array(Int) = Array(1, 3, 5, 7, 9)
c) first () functie gebruikt om het eerste element van de dataset weer te geven:
scala> value.first()
Output:
res4: Int = 1
d) de functie take (n) geeft de eerste n-elementen van de array weer:
scala> value.take(3)
Output:
res6: Array(Int) = Array(1, 3, 5)
e) de functie takeSample (withReplacement, num, (seed)) geeft een willekeurige reeks "num" -elementen weer waarbij de seed voor de random number generator is.
scala> value.takeSample(false, 3, System.nanoTime.toInt)
Output:
res8: Array(Int) = Array(3, 1, 7)
f) saveAsTextFile (pad) functie slaat de gegevensset op in het opgegeven pad van de hdfs-locatie
scala> value.saveAsTextFile("/user/valuedir")
g) partities. lengte functie kan worden gebruikt om het aantal partities in de RDD te vinden
scala> value.partitions.length
Output:
res1: Int = 8
RDD-transformaties
Transformatie wordt gebruikt om een nieuwe RDD te vormen van de bestaande. Omdat de ingangen van de RDD onveranderlijk zijn, kan het resultaat dat wordt gevormd bij transformatie een of meer RDD als uitvoer zijn.
Er zijn twee soorten transformaties:
- Smalle transformaties
- Brede transformaties
Smalle transformaties - Elke bovenliggende RDD is verdeeld in verschillende partities en van deze wordt slechts één partitie gebruikt door de onderliggende RDD.
Voorbeeld: map () en filter () zijn de twee basistypen basistransformaties die worden aangeroepen wanneer een actie wordt aangeroepen.
- map (func) functie werkt iteratief op elk van de elementen in de dataset “waarde” om de output RDD te produceren.
Voorbeeld: In dit voorbeeld voegen we de waarde 10 toe aan elk van de elementen van de gegevenssetwaarde en tonen de getransformeerde uitvoer met behulp van de verzamelfunctie.
scala> val mapfunc = value.map(x => x+10)
mapfunc: org.apache.spark.rdd.RDD(Int) = MapPartitionsRDD(3) at map at :28
scala> mapfunc.collect
res2: Array(Int) = Array(11, 13, 15, 17, 19)
filter (func) functie wordt in principe gebruikt om de elementen uit te filteren die voldoen aan een bepaalde voorwaarde gespecificeerd met behulp van de functie.
Voorbeeld: In dit voorbeeld proberen we alle elementen op te halen behalve nummer 2 van de dataset "waarde" en de uitvoer op te halen via de verzamelfunctie.
scala> val fill = value.filter(x => x!=2)
fill: org.apache.spark.rdd.RDD(Int) = MapPartitionsRDD(7) at filter at :28
scala> fill.collect
res8: Array(Int) = Array(4, 6, 8, 10)
Brede transformaties - Een enkelvoudige RDD-partitie wordt gedeeld op de verschillende meervoudige onderliggende RDD-partities.
Voorbeeld: groupbykey en reducebybyey zijn voorbeelden van brede transformaties.
- De functie groupbyKey groepeert de gegevenssetwaarden in sleutel / waarde-paren volgens de sleutelwaarden van een andere RDD. Dit proces houdt in dat shufflen plaatsvindt wanneer de groep per functie de gegevens verzamelt die aan een bepaalde sleutel zijn gekoppeld en deze in een enkel sleutel / waarde-paar opslaat.
Voorbeeld: In dit voorbeeld wijzen we de gehele getallen 5, 6 toe aan de tekenreekswaarde "key" en het gehele getal 8 toegewezen aan "8" die in de uitvoer in hetzelfde sleutel / waarde-formaat worden weergegeven.
scala> val data = spark.sparkContext.parallelize(Array(("key", 5), ("val", 8), ("key", 6)), 3)
data: org.apache.spark.rdd.RDD((String, Int)) = ParallelCollectionRDD(13) at parallelize at :23
scala> val group = data.groupByKey().collect()
group: Array((String, Iterable(Int))) = Array((key, CompactBuffer(5, 6)), (val, CompactBuffer(8)))
scala> group.foreach(println)
(key, CompactBuffer(5, 6))
(val, CompactBuffer(8))
- reduceByKey-functie combineert ook de sleutel / waarde-paren van verschillende RDD's. Het combineert de sleutels en hun respectieve waarden in een enkel element na het uitvoeren van de genoemde transformatie.
Voorbeeld: In dit voorbeeld worden de algemene toetsen van de array "letters" eerst door de functie parallel aan elkaar gezet en wordt elke letter toegewezen met telnummer 10 eraan. De reduceByKey voegt de waarden met vergelijkbare sleutels toe en slaat de variabele waarde2 op. De uitgang wordt vervolgens weergegeven met de verzamelfunctie.
scala> val letters = Array("A", "B", "C", "D", "B", "C", "E", "D")
letters: Array(String) = Array(A, B, C, D, B, C, E, D)
scala> val value2 = spark.sparkContext.parallelize(letters).map(w => (w, 10)).reduceByKey(_+_)
value2: org.apache.spark.rdd.RDD((String, Int)) = ShuffledRDD(20) at reduceByKey at :25
scala> value2.foreach(println)
(C, 20)
(E, 10)
(D, 20)
(B, 20)
(A, 10)
Samen met de bovengenoemde acties, zoals partitioneren naar RDD en het uitvoeren van acties / transformaties daarop, ondersteunt Spark ook caching, wat handig is wanneer dezelfde gegevens recursief worden genoemd.
Met behulp van al deze eigenschappen kan Apache Spark enorme hoeveelheden gegevens verwerken en batchverwerking en streamingverwerking uitvoeren. De in-memory-berekening van Spark is verantwoordelijk voor de extreem snelle verwerking van applicaties. Daarom is Spark de go-to-methode vanwege de veelzijdigheid van programmeren in verschillende talen, gebruiksgemak en integratiemogelijkheden.
Aanbevolen artikelen
Dit is een handleiding voor Spark Shell Commands. Hier bespreken we de verschillende soorten Spark Shell-opdrachten voor verschillende programmeertalen. U kunt ook het volgende artikel bekijken voor meer informatie -
- Shell Scripting Commando's
- Hoe Spark te installeren
- Spark Interview Vragen
- Vonkopdrachten
- Adhoc testen
- Willekeurige nummergenerator in JavaScript
- Handleiding voor de lijst met Unix Shell-opdrachten
- PySpark SQL | Modules en methoden van PySpark SQL
- Voor Loop in Shell Scripting | Hoe werkt lus?
- Batch-scriptopdrachten met voorbeelden
- Volledig overzicht van vonkcomponenten