Guide de l’utilisateur GraphFrames - Scala

Cet article illustre des exemples du Guide de l’utilisateur GraphFrames.

import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.graphframes._

Création de graphFrames

Vous pouvez créer des GraphFrames à partir de DataFrames de sommets et d'arêtes.

• DataFrame de vertex : un DataFrame de vertex doit contenir une colonne spéciale nommée id qui spécifie des ID uniques pour chaque vertex dans le graphique.
• Edge DataFrame : Un DataFrame d'arête doit contenir deux colonnes spéciales : src (ID de sommet source de l'arête) et dst (ID de sommet de destination de l'arête).

Les deux DataFrames peuvent avoir d’autres colonnes arbitraires. Ces colonnes peuvent représenter des attributs de vertex et de bord.

Créer les sommets et les bords

// Vertex DataFrame
val v = spark.createDataFrame(List(
  ("a", "Alice", 34),
  ("b", "Bob", 36),
  ("c", "Charlie", 30),
  ("d", "David", 29),
  ("e", "Esther", 32),
  ("f", "Fanny", 36),
  ("g", "Gabby", 60)
)).toDF("id", "name", "age")
// Edge DataFrame
val e = spark.createDataFrame(List(
  ("a", "b", "friend"),
  ("b", "c", "follow"),
  ("c", "b", "follow"),
  ("f", "c", "follow"),
  ("e", "f", "follow"),
  ("e", "d", "friend"),
  ("d", "a", "friend"),
  ("a", "e", "friend")
)).toDF("src", "dst", "relationship")

Créons un graphique à partir de ces sommets et de ces bords :

val g = GraphFrame(v, e)

// This example graph also comes with the GraphFrames package.
// val g = examples.Graphs.friends

Requêtes de graphe de base et DataFrame

Les GraphFrames fournissent des requêtes graphiques simples, telles que le degré de nœud.

En outre, étant donné que les GraphFrames représentent des graphiques en tant que paires de trames de données de sommet et de périphérie, il est facile de créer des requêtes puissantes directement sur les trames de données de vertex et de périphérie. Ces DataFrames sont disponibles sous forme de sommets et de champs de bord dans graphFrame.

display(g.vertices)

display(g.edges)

Degré entrant des sommets :

display(g.inDegrees)

Degré sortant des sommets :

display(g.outDegrees)

Degré des sommets :

display(g.degrees)

Vous pouvez exécuter des requêtes directement sur le DataFrame des sommets. Par exemple, nous pouvons trouver l’âge de la plus jeune personne dans le graphique :

val youngest = g.vertices.groupBy().min("age")
display(youngest)

De même, vous pouvez exécuter des requêtes sur la trame de données edges. Par exemple, nous allons compter le nombre de relations « suivre » dans le graphique :

val numFollows = g.edges.filter("relationship = 'follow'").count()

Recherche de motifs

Créez des relations plus complexes impliquant des bords et des sommets à l’aide de motifs. La cellule suivante recherche les paires de sommets avec des bords dans les deux directions. Le résultat est un DataFrame, dans lequel les noms de colonnes sont des clés de motif.

Pour plus d’informations sur l’API, consultez le Guide de l’utilisateur GraphFrame .

// Search for pairs of vertices with edges in both directions between them.
val motifs = g.find("(a)-[e]->(b); (b)-[e2]->(a)")
display(motifs)

Étant donné que le résultat est un DataFrame, vous pouvez créer des requêtes plus complexes en plus du motif. Trouvons toutes les relations réciproques dans lesquelles une personne est âgée de plus de 30 ans :

val filtered = motifs.filter("b.age > 30")
display(filtered)

Requêtes avec état

La plupart des requêtes de motif sont sans état et simples à exprimer, comme dans les exemples ci-dessus. Les exemples suivants illustrent des requêtes plus complexes qui maintiennent l’état le long d’un chemin dans le schéma. Exprimez ces requêtes en combinant la recherche de motif GraphFrame avec des filtres sur le résultat, où les filtres utilisent des opérations de séquence pour construire une série de colonnes DataFrame.

Par exemple, supposons que vous souhaitez identifier une chaîne de 4 sommets avec une propriété définie par une séquence de fonctions. Autrement dit, parmi les chaînes de 4 sommets a->b->c->d, identifiez le sous-ensemble de chaînes correspondant à ce filtre complexe :

• Initialisez l’état sur le chemin.
• Mettre à jour l’état en fonction du sommet a.
• Mettre à jour l’état en fonction du vertex b.
• Etc. pour c et d.
• Si l’état final correspond à une condition, le filtre accepte la chaîne.

Les extraits de code suivants illustrent ce processus, où nous identifions les chaînes de 4 sommets afin qu'au moins 2 des 3 arêtes fassent partie des relations « ami ». Dans cet exemple, l’état est le nombre actuel de bords « friend » ; en général, il peut s’agir de n’importe quelle colonne DataFrame.

// Find chains of 4 vertices.
val chain4 = g.find("(a)-[ab]->(b); (b)-[bc]->(c); (c)-[cd]->(d)")

// Query on sequence, with state (cnt)
//  (a) Define method for updating state given the next element of the motif.
def sumFriends(cnt: Column, relationship: Column): Column = {
  when(relationship === "friend", cnt + 1).otherwise(cnt)
}
//  (b) Use sequence operation to apply method to sequence of elements in motif.
//      In this case, the elements are the 3 edges.
val condition = Seq("ab", "bc", "cd").
  foldLeft(lit(0))((cnt, e) => sumFriends(cnt, col(e)("relationship")))
//  (c) Apply filter to DataFrame.
val chainWith2Friends2 = chain4.where(condition >= 2)
display(chainWith2Friends2)

Sous-graphes

GraphFrames fournit des API pour la création de sous-graphiques en filtrant sur les bords et les sommets. Ces filtres peuvent être composés ensemble. Par exemple, le sous-graphique suivant contient uniquement des personnes qui sont amis et qui ont plus de 30 ans.

// Select subgraph of users older than 30, and edges of type "friend"
val g2 = g
  .filterEdges("relationship = 'friend'")
  .filterVertices("age > 30")
  .dropIsolatedVertices()

Filtres triplets complexes

L’exemple suivant montre comment sélectionner un sous-graphique basé sur des filtres triplet qui fonctionnent sur un bord et ses sommets « src » et « dst ». L’extension de cet exemple pour aller au-delà des triplets à l’aide de motifs plus complexes est simple.

// Select subgraph based on edges "e" of type "follow"
// pointing from a younger user "a" to an older user "b".
val paths = g.find("(a)-[e]->(b)")
  .filter("e.relationship = 'follow'")
  .filter("a.age < b.age")
// "paths" contains vertex info. Extract the edges.
val e2 = paths.select("e.src", "e.dst", "e.relationship")
// In Spark 1.5+, the user may simplify this call:
//  val e2 = paths.select("e.*")

// Construct the subgraph
val g2 = GraphFrame(g.vertices, e2)

display(g2.vertices)

display(g2.edges)

Algorithmes de graphe standard

Cette section décrit les algorithmes de graphe standard intégrés à GraphFrames.

Recherche en largeur (BFS)

Recherchez « Esther » pour les utilisateurs âgés de 32 ans < .

val paths: DataFrame = g.bfs.fromExpr("name = 'Esther'").toExpr("age < 32").run()
display(paths)

La recherche peut également limiter les filtres de bord et les longueurs maximales des chemins d’accès.

val filteredPaths = g.bfs.fromExpr("name = 'Esther'").toExpr("age < 32")
  .edgeFilter("relationship != 'friend'")
  .maxPathLength(3)
  .run()
display(filteredPaths)

Composants connectés

Calculez l’appartenance des composants connectés de chaque vertex et retournez un graphique avec chaque vertex affecté à un ID de composant.

val result = g.connectedComponents.run() // doesn't work on Spark 1.4
display(result)

Composants fortement connectés

Calculez le composant fortement connecté (SCC) de chaque vertex et retournez un graphique avec chaque vertex affecté au SCC contenant ce sommet.

val result = g.stronglyConnectedComponents.maxIter(10).run()
display(result.orderBy("component"))

Propagation d’étiquettes

Exécutez l’algorithme de propagation d’étiquettes statiques pour détecter les communautés dans les réseaux.

Chaque nœud du réseau est initialement affecté à sa propre communauté. À chaque super-étape, les nœuds envoient leur affiliation à la communauté à tous les voisins et mettent à jour leur état selon l'affiliation la plus fréquente des messages entrants.

LPA est un algorithme de détection de communauté standard pour les graphiques. Il est peu coûteux de calcul, même si (1) la convergence n’est pas garantie et (2) on peut finir par des solutions triviales (tous les nœuds identifient dans une seule communauté).

val result = g.labelPropagation.maxIter(5).run()
display(result.orderBy("label"))

PageRank

Identifiez les sommets importants dans un graphique en fonction des connexions.

// Run PageRank until convergence to tolerance "tol".
val results = g.pageRank.resetProbability(0.15).tol(0.01).run()
display(results.vertices)

display(results.edges)

// Run PageRank for a fixed number of iterations.
val results2 = g.pageRank.resetProbability(0.15).maxIter(10).run()
display(results2.vertices)

// Run PageRank personalized for vertex "a"
val results3 = g.pageRank.resetProbability(0.15).maxIter(10).sourceId("a").run()
display(results3.vertices)

Chemins les plus courts

Calcule les chemins les plus courts vers l’ensemble spécifié de sommets de référence, où les sommets de référence sont spécifiés par identifiant de sommet.

val paths = g.shortestPaths.landmarks(Seq("a", "d")).run()
display(paths)

Comptage de triangles

Calcule le nombre de triangles passant par chaque sommet.

import org.graphframes.examples
val g: GraphFrame = examples.Graphs.friends  // get example graph

val results = g.triangleCount.run()
results.select("id", "count").show()