Przewodnik użytkownika programu GraphFrames — Scala
W tym artykule przedstawiono przykłady z podręcznika użytkownika programu GraphFrames.
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.graphframes._
Tworzenie ramek programu Graph
Elementy GraphFrame można tworzyć na podstawie wierzchołków i ramek danych krawędzi.
- Ramka danych wierzchołka: ramka danych wierzchołka powinna zawierać specjalną kolumnę o nazwie
id, która określa unikatowe identyfikatory dla każdego wierzchołka na grafie. - Ramka danych krawędzi: ramka danych krawędzi powinna zawierać dwie specjalne kolumny:
src(identyfikator wierzchołka źródłowego krawędzi) idst(identyfikator docelowego wierzchołka krawędzi).
Obie ramki danych mogą mieć dowolne inne kolumny. Te kolumny mogą reprezentować atrybuty wierzchołków i krawędzi.
Tworzenie wierzchołków i krawędzi
// 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")
Utwórzmy wykres na podstawie tych wierzchołków i tych krawędzi:
val g = GraphFrame(v, e)
// This example graph also comes with the GraphFrames package.
// val g = examples.Graphs.friends
Podstawowe zapytania dotyczące grafu i ramki danych
Elementy GraphFrame zapewniają proste zapytania grafu, takie jak stopień węzła.
Ponadto, ponieważ ramki GraphFrame reprezentują grafy jako pary wierzchołków i ramek danych krawędzi, łatwo jest tworzyć zaawansowane zapytania bezpośrednio na wierzchołkach i krawędziach ramek danych. Te ramki danych są dostępne jako wierzchołki i krawędzie pól w ramce graph.
display(g.vertices)
display(g.edges)
Przychodzący stopień wierzchołków:
display(g.inDegrees)
Stopień wychodzący wierzchołków:
display(g.outDegrees)
Stopień wierzchołków:
display(g.degrees)
Zapytania można uruchamiać bezpośrednio na wierzchołkach Ramka danych. Na przykład możemy znaleźć wiek najmłodszej osoby na wykresie:
val youngest = g.vertices.groupBy().min("age")
display(youngest)
Podobnie zapytania można uruchamiać na krawędziach ramki danych. Na przykład zliczmy liczbę relacji "follow" na wykresie:
val numFollows = g.edges.filter("relationship = 'follow'").count()
Znajdowanie motywów
Twórz bardziej złożone relacje obejmujące krawędzie i wierzchołki przy użyciu motywów. Poniższa komórka znajduje pary wierzchołków z krawędziami w obu kierunkach między nimi. Wynikiem jest ramka danych, w której nazwy kolumn są kluczami motywów.
Aby uzyskać więcej informacji na temat interfejsu API, zapoznaj się z przewodnikiem użytkownika programu 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)
Ponieważ wynikiem jest ramka danych, można tworzyć bardziej złożone zapytania na podstawie motywu. Znajdźmy wszystkie wzajemne relacje, w których jedna osoba jest starsza niż 30:
val filtered = motifs.filter("b.age > 30")
display(filtered)
Zapytania stanowe
Większość zapytań motywów jest bezstanowa i prosta do wyrażenia, jak w powyższych przykładach. W następnych przykładach pokazano bardziej złożone zapytania, które niosą stan wzdłuż ścieżki w motywie. Wyrażaj te zapytania, łącząc wyszukiwanie motywu GraphFrame z filtrami w wyniku, gdzie filtry używają operacji sekwencji do konstruowania serii kolumn ramki danych.
Załóżmy na przykład, że chcesz zidentyfikować łańcuch 4 wierzchołków z właściwością zdefiniowaną przez sekwencję funkcji. Oznacza to, że wśród łańcuchów 4 wierzchołków zidentyfikuj podzestaw łańcuchów a->b->c->dpasujących do tego złożonego filtru:
- Zainicjuj stan na ścieżce.
- Zaktualizuj stan na podstawie wierzchołka a.
- Zaktualizuj stan na podstawie wierzchołka b.
- Itp. c i d.
- Jeśli stan końcowy jest zgodny z pewnym warunkiem, filtr akceptuje łańcuch.
Poniższe fragmenty kodu pokazują ten proces, w którym identyfikujemy łańcuchy 4 wierzchołków, takie jak relacje co najmniej 2 z 3 krawędzi. W tym przykładzie stan jest bieżącą liczbą krawędzi "przyjaciela"; ogólnie rzecz biorąc, może to być dowolna kolumna ramki danych.
// 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)
Podgrafy
Element GraphFrames udostępnia interfejsy API do tworzenia podgrafów przez filtrowanie krawędzi i wierzchołków. Te filtry mogą składać się razem. Na przykład poniższa podgraf zawiera tylko osoby, które są przyjaciółmi i które mają ponad 30 lat.
// Select subgraph of users older than 30, and edges of type "friend"
val g2 = g
.filterEdges("relationship = 'friend'")
.filterVertices("age > 30")
.dropIsolatedVertices()
Złożone filtry potrójne
W poniższym przykładzie pokazano, jak wybrać podgraf na podstawie filtrów potrójnych działających na krawędzi i jej wierzchołkach "src" i "dst". Rozszerzenie tego przykładu, aby wykraczać poza trojaczki, używając bardziej złożonych motywów, jest proste.
// 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)
Standardowe algorytmy grafu
W tej sekcji opisano standardowe algorytmy grafu wbudowane w elementy GraphFrame.
Wyszukiwanie według zakresu (BFS)
Wyszukaj ciąg "Esther" dla użytkowników w wieku 32 lat < .
val paths: DataFrame = g.bfs.fromExpr("name = 'Esther'").toExpr("age < 32").run()
display(paths)
Wyszukiwanie może również ograniczać filtry krawędzi i maksymalną długość ścieżki.
val filteredPaths = g.bfs.fromExpr("name = 'Esther'").toExpr("age < 32")
.edgeFilter("relationship != 'friend'")
.maxPathLength(3)
.run()
display(filteredPaths)
Połączone składniki
Oblicz członkostwo połączonego składnika każdego wierzchołka i zwróć wykres z każdym wierzchołkiem przypisanym identyfikatorem składnika.
val result = g.connectedComponents.run() // doesn't work on Spark 1.4
display(result)
Silnie połączone składniki
Oblicz silnie połączony składnik (SCC) każdego wierzchołka i zwróć wykres z każdym wierzchołkiem przypisanym do SCC zawierającego ten wierzchołek.
val result = g.stronglyConnectedComponents.maxIter(10).run()
display(result.orderBy("component"))
Propagacja etykiet
Uruchom statyczny algorytm propagacji etykiet w celu wykrywania społeczności w sieciach.
Każdy węzeł w sieci jest początkowo przypisany do własnej społeczności. Na każdym superkrokcie węzły wysyłają przynależność społeczności do wszystkich sąsiadów i aktualizują swój stan do przynależności społeczności do społeczności przychodzących wiadomości.
LPA to standardowy algorytm wykrywania społeczności dla grafów. Jest niedrogi obliczeniowo, chociaż (1) zbieżność nie jest gwarantowana i (2) można skończyć z trywialnymi rozwiązaniami (wszystkie węzły identyfikują się w jednej społeczności).
val result = g.labelPropagation.maxIter(5).run()
display(result.orderBy("label"))
Pagerank
Zidentyfikuj ważne wierzchołki na wykresie na podstawie połączeń.
// 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)
Najkrótsze ścieżki
Oblicza najkrótsze ścieżki do danego zestawu wierzchołków punktów orientacyjnych, gdzie punkty orientacyjne określają identyfikator wierzchołka.
val paths = g.shortestPaths.landmarks(Seq("a", "d")).run()
display(paths)
Zliczanie trójkątów
Oblicza liczbę trójkątów przechodzących przez każdy wierzchołek.
import org.graphframes.examples
val g: GraphFrame = examples.Graphs.friends // get example graph
val results = g.triangleCount.run()
results.select("id", "count").show()