GraphFrames-gebruikershandleiding - Scala

In dit artikel worden voorbeelden uit de GraphFrames-gebruikershandleiding beschreven.

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

GraphFrames maken

U kunt GraphFrames maken op basis van hoekpunt- en edge-dataframes.

• Vertex DataFrame: een hoekpunt-DataFrame moet een speciale kolom met de naam id bevatten die unieke id's voor elk hoekpunt in de grafiek specificeert.
• Edge DataFrame: Een edge DataFrame moet twee speciale kolommen bevatten: src (bronhoekpunt-id van edge) en dst (doelhoekpunt-id van edge).

Beide DataFrames kunnen willekeurige andere kolommen hebben. Deze kolommen kunnen hoekpunt- en randkenmerken vertegenwoordigen.

De hoekpunten en randen maken

// 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")

Laten we een grafiek maken van deze hoekpunten en deze randen:

val g = GraphFrame(v, e)

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

Eenvoudige grafiek- en DataFrame-query's

GraphFrames bieden eenvoudige grafiekquery's, zoals knooppuntgraad.

Omdat GraphFrames grafieken vertegenwoordigen als paren van hoekpunt- en edge-DataFrames, is het bovendien eenvoudig om krachtige query's rechtstreeks op het hoekpunt en edge-DataFrames te maken. Deze DataFrames zijn beschikbaar als hoekpunten en randenvelden in het GraphFrame.

display(g.vertices)

display(g.edges)

De binnenkomende hoekpunten:

display(g.inDegrees)

De uitgaande mate van de hoekpunten:

display(g.outDegrees)

De mate van de hoekpunten:

display(g.degrees)

U kunt query's rechtstreeks uitvoeren op de hoekpunten DataFrame. We kunnen bijvoorbeeld de leeftijd van de jongste persoon vinden in de grafiek:

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

Op dezelfde manier kunt u query's uitvoeren op de edges DataFrame. Laten we bijvoorbeeld het aantal volgrelaties in de grafiek tellen:

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

Motief zoeken

Bouw complexere relaties met randen en hoekpunten met behulp van motieven. In de volgende cel vindt u de paar hoekpunten met randen in beide richtingen ertussen. Het resultaat is een DataFrame, waarin de kolomnamen motiefsleutels zijn.

Raadpleeg de GraphFrame-gebruikershandleiding voor meer informatie over de API.

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

Omdat het resultaat een DataFrame is, kunt u complexere query's maken op basis van het motief. Laten we alle wederzijdse relaties zoeken waarin één persoon ouder is dan 30:

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

Stateful query's

De meeste motiefquery's zijn staatloos en eenvoudig uit te drukken, zoals in de bovenstaande voorbeelden. De volgende voorbeelden laten complexere query's zien die de status langs een pad in het motief hebben. Druk deze query's uit door GraphFrame-motiefzoekopdrachten te combineren met filters op het resultaat, waarbij de filters sequentiebewerkingen gebruiken om een reeks DataFrame-kolommen samen te stellen.

Stel dat u een keten van vier hoekpunten wilt identificeren met een bepaalde eigenschap die is gedefinieerd door een reeks functies. Dat wil, onder ketens van 4 hoekpunten a->b->c->d, de subset van ketens identificeren die overeenkomen met dit complexe filter:

• Initialiseer de status op het pad.
• Status bijwerken op basis van hoekpunt a.
• Status bijwerken op basis van hoekpunt b.
• Enz. voor c en d.
• Als de uiteindelijke status overeenkomt met een bepaalde voorwaarde, accepteert het filter de keten.

De volgende codefragmenten demonstreren dit proces, waarbij we ketens van 4 hoekpunten identificeren, zodat ten minste 2 van de drie randen 'vriend'-relaties zijn. In dit voorbeeld is de status het huidige aantal 'vriendranden'; over het algemeen kan dit elke DataFrame-kolom zijn.

// 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)

Subgrafieken

GraphFrames biedt API's voor het bouwen van subgrafieken door te filteren op randen en hoekpunten. Deze filters kunnen samen worden samengesteld. De volgende subgrafiek bevat bijvoorbeeld alleen personen die vrienden zijn en die ouder zijn dan 30 jaar.

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

Complexe drievoudige filters

In het volgende voorbeeld ziet u hoe u een subgrafiek selecteert op basis van tripletfilters die op een rand werken en de bijbehorende 'src' en 'dst'-hoekpunten. Het is eenvoudig om dit voorbeeld uit te breiden tot meer drietallen door complexere motieven te gebruiken.

// 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)

Standaardgrafiekalgoritmen

In deze sectie worden de standaardgrafiekalgoritmen beschreven die zijn ingebouwd in GraphFrames.

Breedte-eerste zoekopdracht (BFS)

Zoek vanuit "Esther" naar gebruikers van 32 jaar < .

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

De zoekopdracht kan ook randfilters en maximale padlengten beperken.

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

Verbonden onderdelen

Bereken het lidmaatschap van het verbonden onderdeel van elk hoekpunt en retourneer een grafiek met elk hoekpunt waaraan een onderdeel-id is toegewezen.

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

Sterk verbonden onderdelen

Bereken het sterk verbonden onderdeel (SCC) van elk hoekpunt en retourneer een grafiek met elk hoekpunt dat is toegewezen aan de SCC die dat hoekpunt bevat.

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

Labeldoorgifte

Voer het statische algoritme voor het doorgeven van labels uit voor het detecteren van community's in netwerken.

Elk knooppunt in het netwerk wordt in eerste instantie toegewezen aan een eigen community. Bij elke stap verzenden knooppunten hun community-lidmaatschap naar alle buren en werken ze hun status bij naar de modus communityrelatie van binnenkomende berichten.

LPA is een standaard communitydetectie-algoritme voor grafieken. Het is goedkoop rekenkundig, hoewel (1) convergentie niet wordt gegarandeerd en (2) kan men eindigen met triviale oplossingen (alle knooppunten identificeren zich in één community).

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

Pagerank

Belangrijke hoekpunten in een grafiek identificeren op basis van verbindingen.

// 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)

Kortste paden

Berekent de kortste paden naar de opgegeven set oriëntatiepunthoeken, waarbij oriëntatiepunten worden opgegeven op basis van hoekpunt-id.

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

Driehoeken tellen

Berekent het aantal driehoeken dat door elk hoekpunt gaat.

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

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