GraphFrames felhasználói útmutató – Scala

Ez a cikk példákat mutat be a GraphFrames felhasználói útmutatójából.

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

GraphFrame-ek létrehozása

GraphFrame-eket csúcspontokból és perem adatkeretekből hozhat létre.

• Csúcs adatkeret: A csúcs adatkeretnek tartalmaznia kell egy speciális, id nevű oszlopot, amely egyedi azonosítókat ad meg a gráf egyes csúcsaihoz.
• Edge DataFrame: Az edge DataFrame-nek két speciális oszlopot kell tartalmaznia: src (az él forrás csúcsazonosítója) és dst (az él cél csúcsazonosítója).

Mindkét adatkeret tetszőleges más oszlopokkal rendelkezhet. Ezek az oszlopok csúcs- és élattribútumokat jelölhetnek.

Csúcsokat és éleket hozz létre

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

Hozzunk létre egy gráfot ezekből a csúcsokból és a következő élekből:

val g = GraphFrame(v, e)

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

Alapszintű gráf- és DataFrame-lekérdezések

A GraphFrame-ek egyszerű gráf lekérdezéseket biztosítanak, például csomópontfokot.

Mivel a GraphFrame-ek csúcs- és edge DataFrame-párokként jelölik a gráfokat, könnyen hatékony lekérdezéseket végezhet közvetlenül a csúcspontokon és az él adatkeretein. Ezek a DataFrame-ek csúcsok és élek mezőiként érhetők el a GraphFrame-ben.

display(g.vertices)

display(g.edges)

A csúcspontok bejövő fokszáma:

display(g.inDegrees)

A csúcspontok kimenő foka:

display(g.outDegrees)

A csúcsok fokszáma:

display(g.degrees)

A lekérdezéseket közvetlenül a csúcsok DataFrame-jén futtathatja. A gráfban például megtalálhatjuk a legfiatalabb személy életkorát:

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

Hasonlóképpen lekérdezéseket is futtathat az edges DataFrame-en. Számoljuk meg például a gráf "követő" kapcsolatainak számát:

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

Motívumkeresés

Összetettebb kapcsolatokat hozhat létre élek és csúcsok használatával, motívumok használatával. Az alábbi cella megkeresi a két irányban széllel rendelkező csúcsok párjait. Az eredmény egy DataFrame, amelyben az oszlopnevek motívumkulcsok.

Az API-val kapcsolatos további részletekért tekintse meg a GraphFrame felhasználói útmutató.

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

Mivel az eredmény egy DataFrame, összetettebb lekérdezéseket is létrehozhat a motívumra. Keressük meg azokat a kölcsönös kapcsolatokat, amelyekben egy személy 30 évnél idősebb:

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

Állapotalapú lekérdezések

A legtöbb motívumos lekérdezés állapot nélküli, és egyszerűen kifejezni, mint a fenti példákban. A következő példák összetettebb lekérdezéseket mutatnak be, amelyek állapotot hordoznak a motívum egy útvonalán. Ezeket a lekérdezéseket a GraphFrame-motívumkeresés és az eredmény szűrőinek kombinálásával fejezheti ki, ahol a szűrők sorozatos DataFrame-oszlopok létrehozásához használnak szekvenciaműveleteket.

Tegyük fel például, hogy egy 4 csúcsból álló láncot szeretne azonosítani egy függvénysorozat által meghatározott tulajdonsággal. Vagyis a a->b->c->d-lal jelölt 4 csúcsból álló láncok között azonosítsa a bonyolult szűrő kritériumainak megfelelő láncok részhalmazát.

• Állapot inicializálása az útvonalon.
• Az állapot frissítése az 'a' csúcs alapján.
• Állapot frissítése a b csúcs alapján.
• Stb. c és d esetén.
• Ha a végső állapot megfelel valamilyen feltételnek, a szűrő elfogadja a láncot.

A következő kódrészletek ezt a folyamatot mutatják be, ahol olyan 4 csúcsból álló láncokat azonosítunk, amelyeknél a lánc 3 éle közül legalább 2 "barát" kapcsolat. Ebben a példában az állapot a "barát" élek aktuális száma; általánosságban véve bármelyik DataFrame oszlop lehet.

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

Algráfok

A GraphFrames API-kat biztosít az algráfok létrehozásához az élekre és csúcsokra való szűréssel. Ezek a szűrők összeállhatnak. Az alábbi algráf például csak olyan személyeket tartalmaz, akik barátok és 30 évnél idősebbek.

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

Összetett hármas szűrők

Az alábbi példa bemutatja, hogyan választhat ki egy algráfot a hármas szűrők alapján, amelyek egy élen és annak "src" és "dst" csúcsain működnek. A példa kibővítése összetettebb motívumok használatával a triplákon túlra is egyszerű.

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

Standard gráf algoritmusok

Ez a szakasz a GraphFramesbe beépített szabványos gráf algoritmusokat ismerteti.

Első szintű keresés (BFS)

Az "Esther" kifejezésre történő keresés a < 32 éves felhasználók számára.

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

A keresés korlátozhatja az élszűrőket és az elérési utak maximális hosszát is.

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

Csatlakoztatott összetevők

Számítsa ki az egyes csúcsok csatlakoztatott összetevő-tagságát, és adjon vissza egy gráfot minden egyes csúcshoz hozzárendelt összetevő-azonosítóval.

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

Erősen csatlakoztatott összetevők

Számítsa ki az egyes csúcsok erősen csatlakoztatott összetevőjét (SCC), és adjon vissza egy gráfot az adott csúcsot tartalmazó SCC-hez rendelt csúcsokkal.

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

Címkepropagálás

Futtassa a statikus címkepropagálási algoritmust a hálózatokban lévő közösségek észleléséhez.

A hálózat minden csomópontja kezdetben a saját közösségéhez van rendelve. Minden szuperlépésnél a csomópontok elküldik hovatartozásukat az összes szomszédnak, és úgy frissítik állapotukat, hogy az a bejövő üzenetek leggyakrabban előforduló közösségi hovatartozásává váljon.

Az LPA egy szabványos közösségi észlelési algoritmus a gráfokhoz. Számítási szempontból olcsó, bár (1) a konvergencia nem garantált, és (2) a végén triviális megoldás is adódhat, ahol minden csomópont egy közösségbe kerül.

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

PageRank

Fontos csúcspontok azonosítása egy gráfban kapcsolatok alapján.

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

Legrövidebb útvonalak

Kiszámítja a tájékozódási pontok adott halmazának legrövidebb útvonalait, ahol a tájékozódási pontok csúcsazonosító alapján vannak megadva.

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

Háromszögek számlálása

Kiszámítja az egyes csúcspontokon áthaladó háromszögek számát.

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

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