Qu'est-ce que les commandes Spark Shell?
Le shell Spark est une interface utilisée pour écrire des requêtes ad hoc pour fonctionner et comprendre le comportement d'Apache Spark. Il est appelé le moteur open source de cluster computing qui peut effectuer le traitement en mémoire de données telles que pour l'analyse, l'ETL, l'apprentissage automatique pour d'énormes ensembles de données. Dans cette rubrique, nous allons découvrir les commandes Spark Shell.
Il existe différents types de shell Spark pour différents langages de programmation par exemple:
- spark-shell est écrit en Scala
- pyspark est en Python et
- sparkR pour le langage R
On peut développer son application autonome avec l'aide de Spark. Il est largement utilisé en raison de sa vitesse de calcul ultra-rapide. En effet, il utilise MapReduce pour traiter diverses requêtes et transformations.
Pour exécuter des commandes spark-shell, il faut que Java et Scala soient déjà installés sur le système.
Types de commandes Spark Shell
Les différents types de commandes Spark-shell sont les suivants:
1. Pour vérifier si le Spark est installé et connaître sa version, la commande ci-dessous est utilisée (toutes les commandes ci-dessous doivent être indiquées en commençant par ce symbole «$»)
$ spark-shell
La sortie suivante s'affiche si l'étincelle est installée:
$ spark-shell
SPARK_MAJOR_VERSION est défini sur 2, en utilisant Spark2
Définition du niveau de journalisation par défaut sur «WARN».
Pour ajuster le niveau de journalisation, utilisez sc.setLogLevel (newLevel). Pour SparkR, utilisez setLogLevel (newLevel).
Interface utilisateur Web de contexte Spark disponible à l'adresse http://10.113.59.34:4040
Contexte Spark disponible en tant que «sc» (master = local (*), id d'application = local-1568732886588).
Session Spark disponible en tant que «spark».
Bienvenue à
____ __
/ __ / __ ___ _____ / / __
_ \ \ / _ \ / _ `/ __ / '_ /
/ ___ / .__ / \ _, _ / _ / / _ / \ _ \ version 2.2.0.2.6.3.0-235
/ _ /
Utilisation de Scala version 2.11.8 (VM serveur HotSpot (TM) 64 bits, Java 1.8.0_112)
Tapez des expressions pour les faire évaluer.
Tapez: aide pour plus d'informations.
scala>
2. La structure de données de base de Spark est appelée RDD (Resilient Distributed Datasets) qui contient une collection immuable d'objets pour le calcul distribué des enregistrements. Tous les jeux de données de RDD sont partitionnés logiquement sur plusieurs nœuds d'un cluster.
Un RDD ne peut être créé qu'en lisant à partir d'un système de fichiers local ou en transformant un RDD existant.
a) Pour créer un nouveau RDD, nous utilisons la commande suivante:
scala> val examplefile = sc.textFile("file.txt")
Ici sc est appelé l'objet de SparkContext.
Production:
examplefile: org.apache.spark.rdd.RDD(String) = file.txt MapPartitionsRDD(3) at textFile at :24
b) Un RDD peut être créé via Parallelized Collection comme suit:
scala> val oddnum = Array(1, 3, 5, 7, 9)
Production:
oddnum: Array(Int) = Array(1, 3, 5, 7, 9)
scala> val value = sc.parallelize(oddnum)
Production:
value: org.apache.spark.rdd.RDD(Int) = ParallelCollectionRDD(4) at parallelize at :26
c) Pour créer à partir de RDD existants :
scala> val newRDD = oddnum.map(value => (value * 2))
Production:
newRDD: Array(Int) = Array(2, 6, 10, 14, 18)
3. Il existe deux types d'opérations Spark RDD qui peuvent être effectuées sur les ensembles de données créés:
- Actions
- Transformations
Actions: Il est utilisé pour effectuer certaines opérations requises sur les jeux de données existants. Voici quelques-unes des commandes qui peuvent être utilisées pour effectuer les actions ci-dessous sur les jeux de données créés:
a) fonction count () pour compter le nombre d'éléments dans RDD:
scala> value.count()
Production:
res3: Long = 5
b) fonction collect () pour afficher tous les éléments du tableau:
scala> value.collect()
Production:
res5: Array(Int) = Array(1, 3, 5, 7, 9)
c) fonction first () utilisée pour afficher le premier élément de l'ensemble de données:
scala> value.first()
Production:
res4: Int = 1
d) la fonction take (n) affiche les n premiers éléments du tableau:
scala> value.take(3)
Production:
res6: Array(Int) = Array(1, 3, 5)
e) La fonction takeSample (withReplacement, num, (seed)) affiche un tableau aléatoire d'éléments «num» où la graine est destinée au générateur de nombres aléatoires.
scala> value.takeSample(false, 3, System.nanoTime.toInt)
Production:
res8: Array(Int) = Array(3, 1, 7)
f) La fonction saveAsTextFile (chemin) enregistre l'ensemble de données dans le chemin spécifié de l'emplacement hdfs
scala> value.saveAsTextFile("/user/valuedir")
g) partitions. la fonction de longueur peut être utilisée pour trouver le nombre de partitions dans le RDD
scala> value.partitions.length
Production:
res1: Int = 8
Transformations RDD
La transformation est utilisée pour former un nouveau RDD à partir des existants. Étant donné que les entrées du RDD sont immuables, le résultat formé lors de la transformation peut être un ou plusieurs RDD en sortie.
Il existe deux types de transformations:
- Transformations étroites
- Transformations larges
Transformations étroites - Chaque RDD parent est divisé en plusieurs partitions et parmi celles-ci, une seule partition sera utilisée par le RDD enfant.
Exemple: map () et filter () sont les deux types de base de transformations de base qui sont appelées lorsqu'une action est appelée.
- La fonction map (func) opère sur chacun des éléments de la «valeur» du jeu de données de manière itérative pour produire le RDD de sortie.
Exemple: Dans cet exemple, nous ajoutons la valeur 10 à chacun des éléments de la valeur de l'ensemble de données et affichons la sortie transformée à l'aide de la fonction collect.
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)
La fonction de filtre (func) est essentiellement utilisée pour filtrer les éléments satisfaisant une condition particulière spécifiée à l'aide de la fonction.
Exemple: Dans cet exemple, nous essayons de récupérer tous les éléments à l'exception du numéro 2 de la «valeur» du jeu de données et de récupérer la sortie via la fonction de collecte.
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)
Transformations larges - Une seule partition RDD parent est partagée sur ses différentes partitions RDD enfants multiples.
Exemple: groupbykey et ReducebyKey sont des exemples de transformations étendues.
- La fonction groupbyKey regroupe les valeurs de l'ensemble de données en paires clé-valeur en fonction des valeurs de clé d'un autre RDD. Ce processus implique un brassage qui a lieu lorsque le groupe par fonction collecte les données associées à une clé particulière et les stocke dans une seule paire clé-valeur.
Exemple: Dans cet exemple, nous attribuons les entiers 5, 6 à la valeur de chaîne «clé» et l'entier 8 affecté à «8» qui sont affichés dans le même format de paire clé-valeur dans la sortie.
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))
- La fonction ReduceByKey combine également les paires clé-valeur de différents RDD. Il combine les clés et leurs valeurs respectives en un seul élément après avoir effectué la transformation mentionnée.
Exemple: Dans cet exemple, les clés communes du tableau «lettres» sont d'abord parallélisées par la fonction et chaque lettre est mappée avec le compte 10. Le ReduceByKey ajoutera les valeurs ayant des clés similaires et enregistre dans la variable value2. La sortie est ensuite affichée à l'aide de la fonction de collecte.
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)
En plus des actions susmentionnées comme le partitionnement en RDD et l'exécution d'actions / transformations sur celles-ci, Spark prend également en charge la mise en cache, ce qui est utile lorsque les mêmes données sont appelées de manière récursive.
À l'aide de toutes ces propriétés, Apache Spark peut traiter d'énormes volumes de données et effectuer un traitement par lots et un traitement en continu. Le calcul en mémoire effectué par Spark est responsable du traitement extrêmement rapide des applications. Par conséquent, Spark est la méthode incontournable en raison de sa polyvalence de programmation sur différents langages, de sa facilité d'utilisation et de ses capacités d'intégration.
Articles recommandés
Ceci est un guide des commandes Spark Shell. Nous discutons ici des différents types de commandes Spark Shell pour différents langages de programmation. Vous pouvez également consulter l'article suivant pour en savoir plus -
- Commandes de script shell
- Comment installer Spark
- Questions d'entretiens chez Spark
- Commandes Spark
- Tests ad hoc
- Générateur de nombres aléatoires en JavaScript
- Guide de la liste des commandes du shell Unix
- PySpark SQL | Modules et méthodes de PySpark SQL
- For Loop dans Shell Scripting | Comment fonctionne la boucle?
- Commandes de script par lots avec des exemples
- Présentation complète des composants Spark