Panduan pengguna GraphFrames - Scala

Artikel ini menunjukkan contoh dari Panduan Pengguna GraphFrames.

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

Membuat GraphFrames

Anda dapat membuat GraphFrames dari vertex dan edge DataFrames.

  • Vertex DataFrame: Vertex DataFrame harus berisi kolom khusus bernama id yang menentukan ID unik untuk setiap vertex dalam grafik.
  • Edge DataFrame: DataFrame tepi harus berisi dua kolom khusus: src (ID verteks sumber tepi) dan dst (ID verteks tujuan tepi).

Kedua DataFrame dapat memiliki kolom lain yang semena-mena. Kolom tersebut dapat mewakili atribut vertex dan edge.

Membuat simpul dan tepi

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

Mari kita buat grafik dari simpul ini dan tepi ini:

val g = GraphFrame(v, e)

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

Kueri grafik dasar dan DataFrame

GraphFrames menyediakan kueri grafik sederhana, seperti derajat simpul.

Selain itu, karena GraphFrames mewakili grafik sebagai pasangan vertex dan edge DataFrames, mudah untuk membuat kueri canggih langsung di vertex dan edge DataFrames. DataFrame tersebut tersedia sebagai simpul dan bidang tepi di GraphFrame.

display(g.vertices)

display(g.edges)

Derajat masuk dari simpul-simpul tersebut:

display(g.inDegrees)

Tingkat keluar dari simpul:

display(g.outDegrees)

Tingkat simpul:

display(g.degrees)

Anda dapat menjalankan kueri langsung di simpul DataFrame. Misalnya, kita dapat menemukan usia orang termuda dalam grafik:

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

Demikian juga, Anda dapat menjalankan kueri di edges DataFrame. Misalnya, mari kita hitung jumlah hubungan 'ikuti' dalam grafik:

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

Temuan motif

Bangun hubungan yang lebih kompleks yang melibatkan tepi dan simpul menggunakan motif. Sel berikut menemukan pasangan simpul dengan tepi di kedua arah di antaranya. Hasilnya adalah DataFrame, di mana nama kolom adalah kunci motif.

Lihat panduan pengguna GraphFrame untuk detail selengkapnya tentang 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)

Karena hasilnya adalah DataFrame, Anda dapat membangun kueri yang lebih kompleks di atas motif. Mari kita temukan semua hubungan timbal balik di mana satu orang lebih tua dari 30:

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

Kueri dengan status

Sebagian besar kueri motif tanpa status dan sederhana untuk diekspresikan, seperti pada contoh di atas. Contoh berikutnya menunjukkan kueri yang lebih kompleks yang membawa status di sepanjang jalur dalam motif. Ekspreskan kueri ini dengan menggabungkan temuan motif GraphFrame dengan filter pada hasilnya, di mana filter menggunakan operasi urutan untuk membuat serangkaian kolom DataFrame.

Misalnya, Anda ingin mengidentifikasi rantai 4 simpul dengan beberapa properti yang ditentukan oleh urutan fungsi. Artinya, di antara rantai yang terdiri dari 4 simpul a->b->c->d, identifikasi subset rantai yang cocok dengan filter kompleks ini:

  • Menginisialisasi keadaan pada jalur.
  • Perbarui keadaan berdasarkan simpul a.
  • Perbarui keadaan berdasarkan simpul b.
  • Dan lain-lain untuk c dan d.
  • Jika status akhir cocok dengan beberapa kondisi, maka filter menerima rantai.

Cuplikan kode berikut menunjukkan proses ini, di mana kami mengidentifikasi rantai 4 simpul sedemikian rupa sehingga setidaknya 2 dari 3 sisi adalah relasi "teman". Dalam contoh ini, statusnya adalah jumlah tepi "teman" saat ini; secara umum, bisa jadi kolom DataFrame apa pun.

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

Subgrafik

GraphFrames menyediakan API untuk membangun subgraf dengan memfilter tepi dan simpul. Filter ini dapat digabungkan bersama. Misalnya, subgraf berikut hanya berisi orang-orang yang berteman dan berusia lebih dari 30 tahun.

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

Triplet filter yang kompleks

Contoh berikut menunjukkan cara memilih subgraf berdasarkan filter triplet yang beroperasi pada sebuah tepi dan simpul-simpul "src" dan "dst". Memperluas contoh ini untuk melampaui tiga serangkai dengan menggunakan motif yang lebih kompleks itu cukup mudah.

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

Algoritma graf standar

Bagian ini menjelaskan algoritma grafik standar yang dibangun ke dalam GraphFrames.

Pencarian yang mengutamakan luas (BFS)

Cari berdasarkan "Esther" untuk pengguna berusia < 32.

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

Pencarian juga dapat membatasi filter tepi dan panjang jalur maksimum.

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

Komponen yang tersambung

Hitung keanggotaan komponen terhubung dari setiap titik dan mengembalikan grafik dengan setiap titik yang memiliki ID komponen.

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

Komponen terhubung kuat

Hitung komponen yang terhubung kuat (SCC) dari setiap simpul dan kembalikan grafik dengan setiap simpul yang ditugaskan ke SCC yang berisi simpul tersebut.

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

Penyebaran label

Jalankan Algoritma Propagasi Label statis untuk mendeteksi komunitas di jaringan.

Setiap simpul dalam jaringan awalnya ditetapkan ke komunitasnya sendiri. Pada setiap superstep, simpul mengirim afiliasi komunitas mereka ke semua tetangga dan memperbarui status mereka ke afiliasi komunitas mode dari pesan masuk.

LPA adalah algoritma deteksi komunitas standar untuk grafik. Ini murah secara komputasi, meskipun (1) konvergensi tidak dijamin dan (2) dapat menghasilkan solusi yang tidak signifikan (semua node teridentifikasi dalam satu komunitas).

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

PageRank (Peringkat Halaman)

Identifikasi simpul penting dalam grafik berdasarkan koneksi.

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

Jalur terpendek

Menghitung jalur terpendek ke set simpul tengara yang diberikan, di mana tengara ditentukan oleh ID simpul.

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

Perhitungan segitiga

Menghitung jumlah segitiga yang melewati setiap puncak.

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

val results = g.triangleCount.run()
results.select("id", "count").show()
  • Last updated on