Trang 1
De nombreux algorithmes de calculs parallèles dans les graphes GraphX Camelia Constantin Prénom.Nom@lip6.fr basé sur la présentation de J. Gonzalez 2 Exemple de calcul de PageRank: Les Graphes sont au centre de l'analyse de données Web Hyperlinks PageRank Top 20 Pages Text Title PR Raw Table Wikipedia Title Body Term-Doc Topic Model < /< / >< / Graph (LDA) Word Topics Rang de >> Wor XML d Topic l'util. i Somme pondérée des rangs des voisins Discussion Community User Community Editor Graph Detection Community Topic Table User Com TopicCom •Les calculs des mises à jour des rangs peuvent être fait en User Disc. . . parallèle •On itère jusqu'à la convergence Les mêmes données peuvent avoir différentes “vues” table ou “vues” graphe (souvent utile de changer entre les deux vues) 3
Trang 2
Systèmes orientés graphe pour chaque vue L'abstraction Graph-Parallel Data-parallel Graph-Parallel Modèle /Modèle / TableTable Graphe de propriétés Alg.Alg. Graphe de propriétés Etat Etat RowRow RowRow RésultatRésultat RowRow Le calcul dépendLe calcul dépend RowRow uniquement desuniquement des voisinsvoisins 6 L’abstraction Graph-Parallel Systèmes Graph-Parallel Un programme dé2ni par l’utilisateur s’exécute sur chaque sommet Le graphe contraint les interactions le long des arêtes: en utilisant des messages (e.g. Pregel [PODC’09, SIGMOD’10]) ou via des états partagés (e.g., GraphLab [UAI’10,VLDB’12]) oogle Proposent des APIs spécialisées pour simplifier la programmation sur des graphes. Nouvelles techniques de partitionnement du graphe, restriction des types d'opérations qui peuvent être utilisées. Exploitent la structure du graphe pour obtenir des gains en performance de plusieurs ordres de magnitude comparé aux systèmes de données parallèles (Data-Parallel) plus génériques. Inconvénients: difficile d'exprimer les différentes étapes d'un pipeline de traitement sur des graphes (contruire/modifier le graphe, calculs sur plusieurs graphes) Parallélisme: lance plusieurs programmes d’arêtes simultannément 7 8
Trang 3
L'abstraction Pregel Illustration Trouver le max des id des noeuds Exemple: calcul de PageRank sur le graphe Pipeline d'analyse de graphes de Live-Journal Preprocessing Compute Post Proc. < / >< / >< / > XML Raw ETL Slice Compute Analyze Data Initial Subgraph PageRank Top GraphLab est 60x plus rapide que Hadoop Graph Users GraphLab is 16x plus rapide que Spark Repeat 12
Trang 4
Inef2cacité Dif2cultés pour la programmation et l’utilisation Les utilisateurs doivent apprendre, déployer, D’importants déplacements de données et de duplications à et gérer de multiples systèmes travers le réseau et le système de 2chiers < / >< / > < / > XML Conduit à des interfaces compliquées à implanter et souvent complexes à utiliser, est particulièrement inef2cace HDFHDF HDFHDF HDFHDF HDFHDF SS SS SS SS Ré-utilisation limitée de structures de données internes d’une étape à l’autre 13 14 L’approche uni2ée GraphX GraphX (Lignes de Code) Nouvelle API New System Atténue la distinction entre Combine les systèmes Data- les Tables et les Graphes Parallel et Graph-Parallel PageRank Connected Shortest SVD ALS K-core (5) Comp. (10) Path (10) (40) (40) (51) Triangle Count LDA (120) (45) Pregel (28) + GraphLab (50) Permet aux utilisateurs: GraphX (3575) d’exprimer facilement et ef2cacement le pipeline entier de l’analyse de graphe. Spark de voir les données à la fois comme collections (RDD) et comme graphe sans déplacement/duplication
Trang 5
GraphX: avantages Un exemple de pipeline Raw Wikipedia Hyperlinks PageRank Top 20 Pages Les abstractions Pregel et GraphLab < / >< / > < / > peuvent être réalisées avec les opérateurs HDFSHDFS HDFSHDFS XML GraphX en moins de 50 lignes de code! Spark PreprocessSpark Preprocess ComputeCompute Spark Post.Spark Post. En composant ces opérateurs on peut 605 construire les pipelines entiers d’analyses de 375 graphe. Le temps de traitement de GraphX pour tout le pipeline est plus rapide 17 que Spark+GraphLab Comparaisons des performances GraphX passe à l’échelle sur de plus grands graphes Live-Journal: 69 Millions d’arêtes Graphe Twitter: 1.5 milliard d’arêtes GraphX est environ 2x plus lent que GraphLab GraphX est environ 3x plus lent que GraphLab » Surcoûts liés à Scala + Java: Lambdas, GC time, … » Pas de parallélisation de la mémoire partgée: 2x plus de comm.
Trang 6
Différentes vues Voir un Graphe comme une Table(RDD) Table des sommets Les Tables et les Graphes sont des vues composables des mêmes Graphe de propriétés Id Propriété (V) données physiques Rxin (Stu., Berk.) R F Jegonzal (PstDoc, Berk.) Franklin (Prof., Berk) IstoicaTable des arêtes(Prof., Berk) Vue Table ReprésentationReprésentation Vue Graphe Src_Id Dst_Id Propriété (E) uni2ée GraphX J I uni2ée GraphX rxin jegonzal Friend franklin rxin Advisor Chaque vue a ses propres opérateurs qui exploitent la sémantique de la istoica franklin Coworker vue pour obtenir une exécution ef2cace franklin jegonzal PI Gestion du graphe Le graphe de propriétés Comme les RDD, les graphes de propriétés sont: Multigraphe dirigé avec des objets définis par l’utilisateur attachés à chaque arête et à Non-modifiable: les changements de valeurs ou chaque sommet de la structure du graphe se font en produisant un nouveau graphe Multi-graphe signifie qu’il peut y avoir plusieurs arêtes partageant la même source Distribués: le graphe est partitionné en utilisant et destination (dans le cas où on veut un ensemble d’heuristiques pour le modéliser plusieurs relations entre nœuds) partitionnement des noeuds Chaque sommet a une clé unique VertexID Résistant aux pannes: comme avec RDD, de 64 bits (VertexId) chaque partition sur le graphe peut être recréé Chaque arête a l’ID du sommet source et sur une autre machine pour la tolérance aux l'ID du sommet destination pannes
Trang 7
Le graphe de propriétés Exemple de graphe de propriétés Le graphe de propriétés correspond à deux RDD : Exemple du graphe de propriétés des collaborateurs du VertexRDD[VD], version optimisée de projet GraphX: RDD[(VertexID,VD)], pour les sommets Les sommets contiennent le nom et la fonction EdgeRDD[ED] versions optimisée de Les arcs annotés avec la nature de la collaboration RDD[Edge[ED]], pour les arcs VD et ED : types des objets associés aux sommets/arcs VertexRDD[VD] et EdgeRDD[ED] fournissent des fonctionnalités supplémentaires pour les calculs de graphe Graphe obtenu: Opérateurs de Tables Construction du graphe type VertexId = Long Les opérateurs de table (RDD) sont hérités de Spark: map reduce sample //Créer une RDD pour les sommets val vertices :RDD[(VertexId,(String, String))] = sc.parallelize(Array((3L, filter count take ("rxin","student")),(7L,("jgonzal","postdoc")),(5L,("franklin","prof")),(2L, ("istoica", "prof")))) groupBy fold first class Edge[ED] (val srcId : VertexId, val dstId : VertexId, val attr : ED) sort reduceByKey partitionBy //Créer une RDD pour les arcs union groupByKey mapWith val edges :RDD[Edge[String]] = sc.parallelize(Array(Edge(3L,7L, "collab"),Edge(5L, 3L,"advisor" ), Edge(2L,5L, "colleague"), join cogroup pipe Edge(5L,7L,"pi"))) leftOuterJoin cross save val graph = Graph(vertices, edges) rightOuterJoin zip ... Le graphe obtenu a la signature : Graph[(String, String), String] 28
Trang 8
Opérateurs de Graphe Utilisation des vues class Graph [ V, E ] { Décomposition du graphe en vue sommets ou vue arêtes avec def Graph(vertices: Table[ (Id, V) ], graph.vertices et graph.edges edges: Table[ (Id, Id, E) ]) // Vues ----------------- val vertices: VertexRDD[VD] val edges: EdgeRDD[ED] val triplets: RDD[EdgeTriplet[VD, ED]] // Transformations ------------------------------ def reverse: Graph[V, E] Résultats : 1, 1 def subgraph(pV: (Id, V) => Boolean, pE: Edge[V,E] => Boolean): Graph[V,E] graph.vertices retourne un VertexRDD[(String,String)] qui étend def mapV(m: (Id, V) => T ): Graph[T,E] RDD[(VertexID,(String,String))] => on peut utiliser le case pour def mapE(m: Edge[V,E] => T ): Graph[V,T] décomposer le tuple (idem pour les arêtes) // Joins ---------------------------------------- def joinV(tbl: Table [(Id, T)]): Graph[(V, T), E ] EdgeRDD[String] RDD[Edge[String]] def joinE(tbl: Table [(Id, Id, T)]): Graph[V, (E, T)] graph.edges retourne ( ) => // Computation ---------------------------------- on pouvait également utiliser case comme suit: def aggregateMessages(sendMsg: EdgeContext[VD, ED, Msg] => Unit, mergeMsg: (Msg, Msg) => Msg, tripletFields: TripletFields = TripletFields.All): VertexRDD[Msg] } 29 Vue triplets Vue triplets En plus de la vue sommets et la vue arêtes, il existe une vue triplets Exemple : afficher les relations entre les utilisateurs : RDD[EdgeTriplet[VD, ED]] La classe EdgeTriplet étend la classe Edge en ajoutant srcAttr et dstAttr contenant les propriétés des noeuds source/destination Résultat : rxin is the collab of jgonzal franklin is the advisor of rxin istoica is the colleague of franklin franklin is the pi of jgonzal
Trang 9
Opérateurs d’information Exemple(1) Calculer le degré entrant de chaque sommet : val graph: Graph[(String, String), String] // Utiliser l'opérateur GraphOps.inDegrees val inDegrees: VertexRDD[Int] = graph.inDegrees inDegrees.collect.foreach(println(_)) Résultat : (3,1) (5,1) (7,2) Exemple(2) Opérateurs de transformation(1) Calculer le degré entrant/sortant/total maximum de chaque sommet : Résultat: mapXX : produit un nouveau graphe de même structure avec XX modifié par la fonction de l’utilisateur map maxDegrees : (5,3) maxInDegree : (7,2) La structure n’est pas affectée (le graphe résultat réutilise les index maxOutDegree : (5,2) structurels du graphe d'origine) Ces opérateurs sont souvent utilisés pour initialiser le graphe pour un calcul ou enlever des propriétés inutiles
Trang 10
Exemple : mapTriplets Opérateurs sur structure Créer un nouveau graphe où chaque arc est étiqueté avec un poids val outputGraph: Graph[Double, Double] = graph.mapTriplets(triplet => 1.0 / triplet.srcAttr).mapVertices((id, _) => 1.0) Opérateurs sur structure (2) Exemple : subgraph (1) //RDD pour les sommets reverse: retourne un nouveau graphe en val users: RDD[(VertexId, (String, String))] = inversant la direction des arêtes sc.parallelize(Array((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),(5L, ("franklin", "prof")), (2L, ("istoica", "prof")),(4L, ("peter", "student")))) subgraph: prend des prédicats sur sommets et arêtes et retourne le graphe contenant les sommets //RDD pour les arcs satisfaisants les prédicats et reliés par les arêtes val relationships: RDD[Edge[String]] = sc.parallelize(Array(Edge(3L, 7L, "collab"), satisfaisant les prédicats (si l'on veut restreindre à Edge(5L, 3L, "advisor"), Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi"),Edge(4L, 0L, "student"), Edge(5L, 0L, "colleague"))) certains nœuds/arcs) mask: retourne un sous-graphe correspondant à // Définir un utilisateur par défaut pour les arcs dont un des sommets n'est pas défini l’intersection d’un graphe donné et d’un graphe- val defaultUser = ("John Doe", "Missing") masque groupEdges: pour un multi-graphe, fusionne les // Construire le graphe différentes arêtes entre 2 sommets en une seule val graph = Graph(users, relationships, defaultUser)
Trang 11
Exemple : subgraph (2) Exemple : subgraph (3) val validGraph = graph.subgraph(vpred = (id, attr) => attr._2 != "Missing") Exemple : subgraph (4) Exemple : mask et subgraph validGraph.vertices.collect.foreach(println(_)) Calculer les composantes connexes en utilisant tous les arrêtes, y compris celles Résultat : passant par des nœuds « inconnus » mais ne pas garder ces nœuds dans le résultat. (2,(istoica,prof)) (3,(rxin,student)) (4,(peter,student)) (5,(franklin,prof)) (7,(jgonzal,postdoc)) validGraph.triplets.map( triplet => triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1).collect.foreach(println(_)) Résultat rxin is the collab of jgonzal franklin is the advisor of rxin istoica is the colleague of franklin franklin is the pi of jgonzal
Trang 12
Opérateurs de jointure Exemple : joinVertices Souvent il est nécessaire de joindre des données de collections externes (RDD) avec des graphes val joinRDD:RDD[(VertexId, String)] = sc.parallelize(Array((3L, "étudiant"), (5L, "professeur"))) val joinedGraph = graph.joinVertices(joinRDD)((id, oldVal, newVal) => (oldVal._1, newVal) ) Ex.: on souhaite ajouter d’autres propriétés à un graphe joinedGraph.vertices.collect.foreach(println(_)) existant, ou copier des propriétés d’un graphe à l’autre Résultat : (2,(istoica, prof)) (3,(rxin,étudiant)) (5,(frankiln,professeur)) (7,(jgonzal, pstdoc)) joinVertices: joint les sommets avec le RDD en entrée et les sommets qui n'ont pas de correspondants dans la RDD gardent leur valeur initiale. retourne un nouveau graphe avec les propriétés obtenues en joinVertices ne peut pas changer le type des propriétés des sommets (ex. : toujours (String, appliquant la fonction map aux sommets joignant String)) Exemple : outerJoinVertices Opérateurs d’agrégation Créer un nouveau graphe, à chaque sommet on associe son degré sortant : val degreesRDD : VertexRDD[Int] = graph.outDegrees val joinedGraph: Graph[Int, String] = graph.outerJoinVertices(degreesRDD)((vid, _, degOpt) => degOpt.getOrElse(0)) joinedGraph.vertices.collect.foreach(println(_)) ● rassembler sur les sommets des informations sur leur voisinage Résultat : (2,1) (3,1) (5,2) (7,0) La fonction map est appliquée à tous les sommets, elle prend comme paramètre un type Option. outerjoinVertices peut changer le type des propriétés des sommets (ex. : Int au lieu de (String, String))
Trang 13
Exemple : Exemple : collectNeighbors aggregateMessages rassembler sur chaque sommet ses voisins et leur attributs pas efficaces, essayer d'utiliser aggregateMessages ● aggregateMessage: applique une fonction sendMsg (≈map) définie par l’utilisateur à chaque val resultat:VertexRDD[Array[(VertexId, (String, String))]] = triplet puis utilise mergeMsg (≈reduce) pour agréger ces messages pour le sommet destination graph.collectNeighbors(EdgeDirection.Either) ● EdgeContext : représente une arrête avec les sommets correspondants et permet d'envoyer resultat.collect.foreach(sommet=>(print(sommet._1), des messages aux sommets sommet._2.foreach(print(_)),println())) ● VertexRDD[Msg] : contient les messages agrégés (de type Msg) pour chaque sommet. Les Résultat : sommets qui n'ont pas reçu de message ne sont pas inclus dans le résultat 2(5,(franklin,prof)) 3(7,(jgonzal,postdoc))(5,(franklin,prof)) ● tripletFields : indique quelles informations de EdgeContext sont accessibles pour sendMsg 5(3,(rxin,student))(2,(istoica,prof))(7,(jgonzal,postdoc)) (autres valeurs possibles : Dst, Src, EdgeOnly, None) 7(3,(rxin,student))(5,(franklin,prof)) Exemple VertexRDD 23 42 Calculer le nombre de followers plus âgés pour VertexRDD[VD] : B C représente un ensemble de nœuds, chacun ayant un attribut de type VD. chaque sommet et la somme de leurs âges chaque VertexID doit être unique, n'apparaît pas explicitement les attributs des nœuds sont stockés dans un hash-map => permet de faire les 30 jointures en temps constants entre deux VertexRDD dérivées à partir de la même VertexRDD A D 19 E 75 ● olderFollowers.collect.foreach(println(_)) Résultats : (A,(1,42.0)) F (C,(1,75.0)) 16 (F,(1,30.0)) 51
Trang 14
Exemple: moyenne des âges des EdgeRDD followers plus âgés Les arcs sont organisés en blocks, partitionnés avec une des stratégies 23 42 (CanonicalRandomVertexCut, EdgePartition1D, EdgePartition2D, B C RandomVertexCut) Les attributs sont stockés séparément de la structure d'adjacence afin de pouvoir les changer facilement 30 Trois fonctions additionnelles, en plus des fonctions hérités de la classe RDD : A ● avgAgeOfOlderFollowers.collect.foreach(println(_)) D 19 E 75 Résultats : (A, 42.0) (C, 75.0) (F, 30.0) F 16 Autres opérateurs API Pregel ● Algorithmes itératifs qui calculent les propriétés des sommets en fonction des propriétés de leur voisins ● Arrêt du calcul : lorsqu'il n'y a plus de messages (à chaque étape les sommets qui n'ont pas reçu de message n'envoient pas de messages) ● vprog : fonction exécutée par chaque sommet pour agréger ses propriétés avec les messages reçus à chaque étape ● sendMsg :pour envoyer un message à chaque voisin ● mergeMsg : fonction pour agréger les messages reçus
Trang 15
Utilisation de PageRank en Plus courts chemins GraphX 2 2 3 1 1 7 8 4 4 3 5 5 6 6 Résultats : (1,0.0) (2,12.0) (3,14.0) (4,8.0) (5,9.0) (6,3.0) Joindre les Sommets et les Arêtes Map Reduce Triplets L’opérateur triplets joint les sommets et les arêtes: Map-Reduce pour chaque sommet B C Sommets Triplets Arêtes AA AA BB mapF( A B ) A BB A C 1 A A C C B C B C mapF( ) A C A 2 D C D C D D E L’opérateur mrTriplets somme les triplets adjacents. reduceF( A , A ) A SELECT t.dstId, reduceUDF( mapUDF(t) ) AS sum 1 2 F FROM triplets AS t GROUPBY t.dstId 60
Trang 16
Conclusion et Observations Vues spéci$ques à un domaine: Tables et Graphes » Les tables et les graphes sont les objects de base composables Programmer avec GraphX » opérateurs spécialisés qui exploitent la sémantique des vues Un système unique qui couvre e,cacement le pipeline » minimise les mouvements et la duplication de données » Plus besoin d’apprendre et gérer différents systèmes http://spark.apache.org/docs/latest/graphx-programming-guide.html Les graphes vus par l’oeil des BD » Graph-Parallel Pattern Triplet joints en algèbre relationnelle » Systèmes de graphe optimisations de jointures distribuées 61 Pour démarrer TME il faut d’abord importer GraphX dans la console Exercice 1: Construire un graphe à partir des fichiers Spark: facebook_edges_prop.csv et facebook_users_prop.csv (voir la section "Graph Builders"). Le fichier facebook_edges_prop.csv contient sur chaque ligne les import org.apache.spark.graphx._ informations suivantes pour un arc: import org.apache.spark.rdd.RDD source, destination, type_relation, nombre_messages_échangés Le fichier facebook_users_prop.csv contient les informations suivantes sur chaque utilisateur: utilisateur, prénom, nom, âge
Trang 17
TME TME Le type de la structure Spark qui stocke les sommets est : Afficher les prénoms des amis de Kendall. On considère le graphe non-dirigé. org.apache.spark.rdd.RDD[(org.apache.spark.graphx.VertexId, (String, String, Int))] Le type de la structure Spark qui stocke les arêtes est : graph.triplets.filter{t =>t.srcAttr._1 =="Kendall" || t.dstAttr._1 == org.apache.spark.rdd.RDD[org.apache.spark.graphx.Edge[(String, Int)]] "Kendall"}.map(t=>if(t.srcAttr._1=="Kendall") t.dstAttr._1 else Le type de la structure Spark qui stocke le graphe est : t.srcAttr._1).collect.foreach(u=>println(u)) org.apache.spark.graphx.Graph[(String, String, Int),(String, Int)] Réponse TME Afficher pour l'utilisateur 'Kendall' son nom et son âge Il faut source, destination, type_relation, nombre_messages_échangés (utiliser la vue graph.vertices). Code var lines = sc.textFile("facebook_edges_prop.csv", 4) val edgesList:RDD[Edge[(String, Int)]] = lines.map{s=> val parts = s.split(",") graph.vertices.filter{case(id,(p,n,a)) => Edge(parts(0).toLong, parts(1).toLong, (parts(2), parts(3).toInt)) }.distinct() p=="Kendall"}.collect.foreach(u=>println(u._1+" var lines = sc.textFile("facebook_users_prop.csv", 4) "+u._2._2+" "+u._2._3)) val vertexList:RDD[(VertexId,(String, String, Int))] = lines.map{s=> val parts = s.split(",") (parts(0).toLong, (parts(1), parts(2), parts(3).toInt)) }.distinct() val graph = Graph.apply(vertexList, edgesList)
Hiển thị 17 trang đầu. Tải tài liệu để xem đầy đủ.
Bình luận (0)
Chưa có bình luận nào. Hãy là người đầu tiên!