Was sind Spark-Shell-Befehle?
Die Spark-Shell ist eine Schnittstelle, mit der Ad-hoc-Abfragen geschrieben werden, um das Verhalten von Apache Spark zu verstehen. Es wird als Cluster-Computing-Open-Source-Engine bezeichnet, die In-Memory-Verarbeitung von Daten durchführen kann, z. B. für Analytics, ETL und maschinelles Lernen für große Datenmengen. In diesem Thema lernen wir die Spark-Shell-Befehle kennen.
Es gibt verschiedene Arten von Spark-Shell für verschiedene Programmiersprachen, zum Beispiel:
- Funken-Muschel ist in Scala geschrieben
- Pyspark ist in Python und
- sparkR für R language
Man kann seine eigenständige Anwendung mit Hilfe von Spark entwickeln. Es ist wegen seiner superschnellen Rechengeschwindigkeit weit verbreitet. Dies liegt daran, dass MapReduce zum Verarbeiten verschiedener Abfragen und Transformationen verwendet wird.
Um Spark-Shell-Befehle ausführen zu können, müssen Java und Scala bereits auf dem System installiert sein.
Arten von Spark-Shell-Befehlen
Die verschiedenen Arten von Spark-Shell-Befehlen lauten wie folgt:
1. Um zu überprüfen, ob der Spark installiert ist und um seine Version zu kennen, wird der folgende Befehl verwendet.
$ spark-shell
Die folgende Ausgabe wird angezeigt, wenn der Funke installiert ist:
$ spark-shell
SPARK_MAJOR_VERSION wird mit Spark2 auf 2 gesetzt
Festlegen der Standardprotokollstufe auf "WARN".
Verwenden Sie zum Anpassen der Protokollierungsstufe sc.setLogLevel (newLevel). Verwenden Sie für SparkR setLogLevel (newLevel).
Spark-Kontext-Web-Benutzeroberfläche verfügbar unter http://10.113.59.34:4040
Spark-Kontext verfügbar als 'sc' (master = local (*), App-ID = local-1568732886588).
Funksitzung als 'Funke' verfügbar.
Willkommen zu
____ __
/ __ / __ ___ _____ / / __
_ \ \ / _ \ / _ `/ __ / '_ /
/ ___ / .__ / \ _, _ / _ / / _ / \ _ \ version 2.2.0.2.6.3.0-235
/ _ /
Verwenden von Scala Version 2.11.8 (Java HotSpot (TM) 64-Bit-Server-VM, Java 1.8.0_112)
Geben Sie Ausdrücke ein, damit sie ausgewertet werden.
Geben Sie: help ein, um weitere Informationen zu erhalten.
scala>
2. Die grundlegende Datenstruktur von Spark wird als RDD (Resilient Distributed Datasets) bezeichnet. Sie enthält eine unveränderliche Sammlung von Objekten für die verteilte Berechnung von Datensätzen. Alle RDD-Datasets sind logisch auf mehrere Knoten eines Clusters verteilt.
Eine RDD kann nur durch Lesen aus einem lokalen Dateisystem oder durch Transformieren einer vorhandenen RDD erstellt werden.
a) Um eine neue RDD zu erstellen, verwenden wir den folgenden Befehl:
scala> val examplefile = sc.textFile("file.txt")
Hier heißt sc das Objekt von SparkContext.
Ausgabe:
examplefile: org.apache.spark.rdd.RDD(String) = file.txt MapPartitionsRDD(3) at textFile at :24
b) Ein RDD kann wie folgt über Parallelized Collection erstellt werden:
scala> val oddnum = Array(1, 3, 5, 7, 9)
Ausgabe:
oddnum: Array(Int) = Array(1, 3, 5, 7, 9)
scala> val value = sc.parallelize(oddnum)
Ausgabe:
value: org.apache.spark.rdd.RDD(Int) = ParallelCollectionRDD(4) at parallelize at :26
c) So erstellen Sie aus vorhandenen RDDs :
scala> val newRDD = oddnum.map(value => (value * 2))
Ausgabe:
newRDD: Array(Int) = Array(2, 6, 10, 14, 18)
3. Es gibt zwei Arten von Spark-RDD-Vorgängen, die für die erstellten Datensätze ausgeführt werden können:
- Aktionen
- Transformationen
Aktionen: Hiermit werden bestimmte erforderliche Vorgänge für die vorhandenen Datasets ausgeführt. Im Folgenden sind einige der Befehle aufgeführt, mit denen die folgenden Aktionen für die erstellten Datensätze ausgeführt werden können:
a) count () - Funktion zum Zählen der Anzahl der Elemente in RDD:
scala> value.count()
Ausgabe:
res3: Long = 5
b) collect () - Funktion zum Anzeigen aller Elemente des Arrays:
scala> value.collect()
Ausgabe:
res5: Array(Int) = Array(1, 3, 5, 7, 9)
c) Funktion first () zur Anzeige des ersten Elements des Datensatzes:
scala> value.first()
Ausgabe:
res4: Int = 1
d) Die Funktion take (n) zeigt die ersten n Elemente des Arrays an:
scala> value.take(3)
Ausgabe:
res6: Array(Int) = Array(1, 3, 5)
e) Die Funktion takeSample (withReplacement, num, (seed)) zeigt ein zufälliges Array von "num" -Elementen an, wobei das seed für den Zufallszahlengenerator ist.
scala> value.takeSample(false, 3, System.nanoTime.toInt)
Ausgabe:
res8: Array(Int) = Array(3, 1, 7)
f) Mit der Funktion saveAsTextFile (Pfad) wird das Dataset im angegebenen Pfad des hdfs-Speicherorts gespeichert
scala> value.saveAsTextFile("/user/valuedir")
g) Partitionen. Mit der Längenfunktion kann die Anzahl der Partitionen im RDD ermittelt werden
scala> value.partitions.length
Ausgabe:
res1: Int = 8
RDD-Transformationen
Die Transformation wird verwendet, um eine neue RDD aus den vorhandenen zu bilden. Da die Eingaben der RDD unveränderlich sind, kann das Ergebnis, das bei der Transformation gebildet wird, eine oder mehrere RDD als Ausgabe sein.
Es gibt zwei Arten von Transformationen:
- Enge Transformationen
- Breite Transformationen
Enge Transformationen - Jede übergeordnete RDD ist in verschiedene Partitionen unterteilt, von denen nur eine von der untergeordneten RDD verwendet wird.
Beispiel: map () und filter () sind die beiden grundlegenden Arten von Transformationen, die beim Aufruf einer Aktion aufgerufen werden.
- Die Funktion map (func) bearbeitet jedes der Elemente im Datensatz "value" iterativ, um die Ausgabe-RDD zu erzeugen.
Beispiel: In diesem Beispiel addieren wir den Wert 10 zu jedem Element des Datensatzwerts und zeigen die transformierte Ausgabe mit Hilfe der Sammelfunktion an.
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)
Die Funktion filter (func) dient im Wesentlichen dazu, die Elemente herauszufiltern, die eine bestimmte Bedingung erfüllen, die mit der Funktion festgelegt wurde.
Beispiel: In diesem Beispiel wird versucht, alle Elemente außer Nummer 2 des Datensatzes „value“ abzurufen und die Ausgabe über die Funktion collect abzurufen.
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)
Weitreichende Transformationen - Eine übergeordnete RDD-Partition wird auf mehreren untergeordneten RDD-Partitionen gemeinsam genutzt.
Beispiel: groupbykey und reductionbyKey sind Beispiele für breite Transformationen.
- Die Funktion groupbyKey gruppiert die Datensatzwerte gemäß den Schlüsselwerten einer anderen RDD in Schlüsselwertpaare. Bei diesem Vorgang wird gemischt, wenn die Funktion group by die einem bestimmten Schlüssel zugeordneten Daten erfasst und in einem einzelnen Schlüssel-Wert-Paar speichert.
Beispiel: In diesem Beispiel weisen wir die Ganzzahlen 5, 6 dem Zeichenfolgenwert „key“ und die Ganzzahl 8 dem Wert „8“ zu, die in der Ausgabe im selben Schlüssel-Wert-Paar-Format angezeigt werden.
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))
- Die Funktion reductByKey kombiniert auch die Schlüssel-Wert-Paare aus verschiedenen RDDs. Es kombiniert die Schlüssel und ihre jeweiligen Werte nach Durchführung der erwähnten Transformation zu einem einzigen Element.
Beispiel: In diesem Beispiel werden die allgemeinen Tasten des Arrays "Buchstaben" zuerst von der Funktion parallelisiert und jedem Buchstaben wird die Anzahl 10 zugeordnet. Der Befehl reduByKey fügt die Werte mit ähnlichen Schlüsseln hinzu und speichert sie in der Variablen value2. Die Ausgabe wird dann mit der Sammelfunktion angezeigt.
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)
Neben den oben genannten Aktionen wie Partitionieren in RDD und Ausführen von Aktionen / Transformationen unterstützt Spark auch das Zwischenspeichern, was hilfreich ist, wenn dieselben Daten rekursiv aufgerufen werden.
Mithilfe all dieser Eigenschaften kann Apache Spark große Datenmengen verarbeiten und eine Stapelverarbeitung sowie eine Streaming-Verarbeitung durchführen. Die von Spark durchgeführte In-Memory-Berechnung ist für die extrem schnelle Verarbeitung von Anträgen verantwortlich. Daher ist Spark aufgrund seiner Vielseitigkeit bei der Programmierung in verschiedenen Sprachen, seiner Benutzerfreundlichkeit und seiner Integrationsmöglichkeiten die bevorzugte Methode.
Empfohlene Artikel
Dies ist eine Anleitung zu Spark Shell-Befehlen. Hier diskutieren wir die verschiedenen Arten von Spark-Shell-Befehlen für verschiedene Programmiersprachen. Sie können auch den folgenden Artikel lesen, um mehr zu erfahren -
- Shell-Scripting-Befehle
- So installieren Sie Spark
- Fragen in Vorstellungsgesprächen bei Spark
- Spark-Befehle
- Ad-hoc-Tests
- Zufallszahlengenerator in JavaScript
- Leitfaden zur Liste der Unix-Shell-Befehle
- PySpark SQL | Module und Methoden von PySpark SQL
- Für Loop in Shell Scripting | Wie funktioniert for loop?
- Stapel-Scripting-Befehle mit Beispielen
- Komplette Übersicht der Spark-Komponenten