Remerciez-le!

Remerciez @Admin pour avoir partagé cet document gratuitement, de la manière la plus simple, en partageant sur les réseaux sociaux.

COURS Préparation à la certification : Big Data Chapitre 5 : Spark Spark Qu’est

COURS Préparation à la certification : Big Data Chapitre 5 : Spark Spark Qu’est que Spark ? ■Apache Spark is a powerful open source processing engine for Hadoop data built around speed, ease of use, and sophisticated analytics. It was originally developed in 2009 in UC Berkeley’s AMPLab, and open sourced in 2010. ■Ce n’est pas une version modifiée de Hadoop. ■C’est un moteur MapReduce plus évolué, plus rapide. o Utilisation de la mémoire pour optimiser les traitements. o Des API’s pour faciliter et optimiser les étapes d’analyses. o Permettre le traitement temps réel des données. ■Compatible avec le système de stockage de Hadoop (HDFS, Hive, SequenceFiles...) 2 Ahlem Ben Younes Spark Spark, plus rapide ■RDD’s (Resilient Distributed DataSet) - Une collection d’objets distribués, mise en cache en mémoire au travers du cluster. - Une API pour manipuler ces objets Operators. - Système de fail over automatique. 3 Ahlem Ben Younes Spark Spark, facile à utiliser .  Une liste d’Operators pour faciliter la manipulation des données au travers des RDD’S. - Map, filter, groupBy, sort, join, leftOuterJoin, ritghtOuterJoin, reduce, count, reduceByKey, groupByKey, first, union, cross, sample, cogroup, take, partionBy, pipe, save...  Compatible avec les langages de programmation fonctionnel Scala, Python, R et Java. 4 Ahlem Ben Younes Spark Spark, un framework analytique ■Shark : Moteur de requête SQL, basé sur le coeur de Spark (in memory), il permet d’exécuter des requêtes hive sans modification 100x plus rapidement. ■MLLib : Library d’algorithmes d’analyses de données (data mining), scalable et optimisé (limite des interactions, RDD’s), k-means, ALS, L2− > L1... ■Spark Streaming : A` opposer aux traitements batch, mise à jour d’un état dynamiquement, piloté par des évènements sur les données event processing (Filtre de Spam, analyse de click, reporting temps réel...) 5 Ahlem Ben Younes Spark Premier exemple Spark Soit un fichier de données de type CSV provenant de http://opendata.paris.fr décrivant des arbres Chaque ligne décrit un arbre : position GPS, arrondissement, genre, espèce, famille, année de plantation, hauteur, circonférence, etc. Le séparateur est ';'. La première ligne contient les titres. Question : On souhaite afficher l’année de plantation (champ n°6) de l’arbre le plus grand (champ n°7). Avec des commandes Unix, ce traitement s’écrirait : Programme pySpark : arbres.py 6 Ahlem Ben Younes spark-submit arbres.py lancer le programme Spark Lancement 7 Ahlem Ben Younes Spark offre plusieurs manières de lancer le programme, dont : • Lancement sur un cluster de Spark Workers spark-submit --master spark://$(hostname -f):7077 \ arbres.py L’option --master de cette commande indique à Spark qu’on doit faire appel au cluster de machines sur lesquelles tournent des Spark Workers. Ce sont des processus clients chargés de faire les calculs distribués pour Spark. • Spark permet aussi de lancer l’exécution sur YARN : spark-submit --master yarn-cluster arbres.py Ce sont les esclaves YARN qui exécutent le programme Spark. Spark Fonction lambda ou fonction nommée ? Le même exemple complet avec des fonctions nommées : 8 Ahlem Ben Younes Éléments de l’API Spark : Principes Données : – RDD : ils représentent des données distribuées modifiées par une transformation, par exemple un map ou un filter. – Variables partagées entre des traitements et distribuées sur le cluster de machines. Calculs : – Transformations : ce sont des fonctions (au sens mathématique) du type : RDD transformation(RDD). Elles créent un nouveau RDD à partir d’un existant. – Actions : ce sont des fonctions qui permettent d’extraire des informations des RDD, par exemple les afficher sur l’écran ou les enregistrer dans un fichier 9 Ahlem Ben Younes Éléments de l’API Spark : Début d’un programme sc représente le contexte Spark. C’est un objet qui possède plusieurs méthodes dont celles qui créent des RDD. lancer le programme, spark-submit essai1.py 10 Ahlem Ben Younes Éléments de l’API Spark : RDD Un RDD est une collection de données abstraite, résultant de la transformation d’un autre RDD ou d’une création à partir de données existantes. Un RDD est distribué, c’est à dire réparti sur plusieurs machines afin de paralléliser les traitements. On peut créer un RDD de deux manières : • Paralléliser une collection • Jeux de données externes Spark peut utiliser de nombreuses sources de données Hadoop : HDFS, HBase. . . et il connait de nombreux types de fichiers. Exemple: donnees = ['veau', 'vache', 'cochon', 'couvée'] RDD = sc.parallelize(donnees) RDD = sc.textFile("hdfs:/share/data.txt") Comme avec MapReduce, chaque ligne du fichier constitue un enregistrement. Les transformations appliquées sur le RDD traiteront chaque ligne séparément. Les lignes du fichier sont distribuées sur différentes machines pour un traitement parallèle. 11 Ahlem Ben Younes Éléments de l’API Spark : Lire et écrire des SequenceFile Certains traitements Spark font appel à la notion de paires (clé,valeur). C’est le cas de l’exemple initial. Les clés permettent par exemple de classer des valeurs dans un certain ordre. Pour stocker efficacement ce genre de RDD, on emploie un SequenceFile. Lecture d’un SequenceFile dans un RDD Cette fonction lit les paires du fichier et crée un RDD : RDD = sc.sequenceFile("hdfs:/share/data1.seq") Écriture d’un RDD dans un SequenceFile Cette méthode enregistre les paires (clé, valeur) du RDD : RDD.saveAsSequenceFile("hdfs:/share/data2.seq") 12 Ahlem Ben Younes Éléments de l’API Spark : Actions Ce sont des méthodes qui s’appliquent à un RDD pour retourner une valeur ou une collection. • liste = RDD.collect() retourne le RDD sous forme d’une liste Python. Attention à la taille si c’est du BigData. • nombre = RDD.count() retourne le nombre d’éléments • premier = RDD.first() retourne le premier élément • premiers = RDD.take(n) retourne les n premiers éléments. Note: il n’y a pas de méthode last pour retourner le ou les derniers éléments. • resultat = RDD.reduce(fonction) applique une fonction d’agrégation (associative) du type fn(a,b)->c • grand = RDD.reduce(lambda a,b: max(a,b)) 13 Ahlem Ben Younes Éléments de l’API Spark : Transformations 14 Ahlem Ben Younes Éléments de l’API Spark : Transformations de type map Chacune de ces méthodes retourne un nouveau RDD à partir de celui qui est concerné (appelé self en Python). • RDD.map(fonction) : chaque appel à la fonction doit retourner une valeur qui est mise dans le RDD sortant. RDD = sc.parallelize([1,2,3,4]) print RDD.map(lambda n: n+1).collect() • RDD.filter(fonction) : la fonction retourne un booléen. Il ne reste du RDD que les éléments pour lesquels la fonction retourne True. RDD = sc.parallelize([1,2,3,4]) print RDD.filter(lambda n: (n%2)==0).collect() • RDD.flatMap(fonction) : chaque appel à la fonction doit retourner une liste (vide ou pas) et toutes ces listes sont concaténées dans le RDD sortant. RDD = sc.parallelize([0,1,2,3]) print RDD.flatMap(lambda n: [n]*n).collect() les résultats de la lambda sont [], [1], [2, 2], [3, 3, 3] et au retour du flatMap on aura [1, 2, 2, 3, 3, 3] 15 Ahlem Ben Younes Éléments de l’API Spark : Transformations ensemblistes Ces transformations regroupent deux RDD, self et celui passé en paramètre.  •RDD.distinct() : retourne un seul exemplaire de chaque élément. RDD = sc.parallelize([1, 2, 3, 4, 6, 5, 4, 3]) print RDD.distinct().collect()  RDD1.union(RDD2) : contrairement à son nom, ça retourne la concaténation et non pas l’union des deux RDD. Rajouter distinct() pour faire une vraie union. RDD1 = sc.parallelize([1,2,3,4]) RDD2 = sc.parallelize([6,5,4,3]) print RDD1.union(RDD2).collect() print RDD1.union(RDD2).distinct().collect()  RDD1.intersection(RDD2) : retourne l’intersection des deux RDD. RDD1 = sc.parallelize([1,2,3,4]) RDD2 = sc.parallelize([6,5,4,3]) print RDD1.intersection(RDD2).collect() 16 Ahlem Ben Younes Éléments de l’API Spark : Transformations sur des paires (clé, valeur) Les transformations suivantes manipulent des RDD dont les éléments sont des paires (clé, valeur)  RDD.groupByKey() : retourne un RDD dont les éléments sont des paires (clé, liste des valeurs ayant cette clé dans le RDD concerné).  RDD.sortByKey(ascending) : retourne un RDD dont les clés sont triées. Mettre True ou False.  RDD.reduceByKey(fonction) : regroupe les valeurs ayant la même clé et leur applique la fonction RDD = sc.parallelize([ (1,"paul"),(2,"anne"),(1,"emile"),(2,"marie"),(1,"victor") ]) print RDD.reduceByKey(lambda a,b: a+"-"+b).collect() retourne [(1, "paul-emile-victor"), (2, "anne-marie")] 17 Ahlem Ben Younes Éléments de l’API Spark : Transformations de type jointure Spark permet de calculer des jointures entre RDD1={(K1,V1). . . } et RDD2={(K2,V2). . . } et partageant des clés K identiques. • RDD1.join(RDD2) : retourne toutes les paires (K, (V1, V2)) lorsque V1 et V2 ont la même clé. • RDD1.leftOuterJoin(RDD2) : retourne les paires (K, (V1, V2)) ou (K, (V1, None)) si (K,V2) manque dans RDD2 • RDD1.rightOuterJoin(RDD2) : retourne les paires (K, (V1, V2)) ou (K, (None, V2)) si (K,V1) manque dans RDD1 • RDD1.fullOuterJoin(RDD2) : retourne toutes les paires (K, (V1, V2)), (K, (V1, None)) ou (K, (None, V2)) 18 Ahlem Ben Younes uploads/Sante/ lecon5-spark.pdf