Spark GraphX屬性圖

Spark GraphX屬性圖

屬性圖是一個(gè)有向多重圖，它帶有連接到每個(gè)頂點(diǎn)和邊的用戶定義的對象。有向多重圖中多個(gè)并行(parallel)的邊共享相同的源和目的地頂點(diǎn)。支持并行邊的能力簡化了建模場景，這個(gè)場景中，相同的頂點(diǎn)存在多種關(guān)系(例如co-worker和friend)。每個(gè)頂點(diǎn)由一個(gè)唯一的64位長的標(biāo)識符（VertexID）作為key。GraphX并沒有對頂點(diǎn)標(biāo)識強(qiáng)加任何排序。同樣，頂點(diǎn)擁有相應(yīng)的源和目的頂點(diǎn)標(biāo)識符。

屬性圖通過vertex(VD)和edge(ED)類型參數(shù)化，這些類型是分別與每個(gè)頂點(diǎn)和邊相關(guān)聯(lián)的對象的類型。

在某些情況下，在相同的圖形中，可能希望頂點(diǎn)擁有不同的屬性類型。這可以通過繼承完成。例如，將用戶和產(chǎn)品建模成一個(gè)二分圖，我們可以用如下方式

class VertexProperty()
case class UserProperty(val name: String) extends VertexProperty
case class ProductProperty(val name: String, val price: Double) extends VertexProperty
// The graph might then have the type:
var graph: Graph[VertexProperty, String] = null

和RDD一樣，屬性圖是不可變的、分布式的、容錯的。圖的值或者結(jié)構(gòu)的改變需要按期望的生成一個(gè)新的圖來實(shí)現(xiàn)。注意，原始圖的大部分都可以在新圖中重用，用來減少這種固有的功能數(shù)據(jù)結(jié)構(gòu)的成本。執(zhí)行者使用一系列頂點(diǎn)分區(qū)試探法來對圖進(jìn)行分區(qū)。如RDD一樣，圖中的每個(gè)分區(qū)可以在發(fā)生故障的情況下被重新創(chuàng)建在不同的機(jī)器上。

邏輯上的屬性圖對應(yīng)于一對類型化的集合(RDD),這個(gè)集合編碼了每一個(gè)頂點(diǎn)和邊的屬性。因此，圖類包含訪問圖中頂點(diǎn)和邊的成員。

class Graph[VD, ED] {
  val vertices: VertexRDD[VD]
  val edges: EdgeRDD[ED]
}

VertexRDD[VD]和EdgeRDD[ED]類分別繼承和優(yōu)化自RDD[(VertexID, VD)]和RDD[Edge[ED]]。VertexRDD[VD]和EdgeRDD[ED]都支持額外的功能來建立在圖計(jì)算和利用內(nèi)部優(yōu)化。

屬性圖的例子

在GraphX項(xiàng)目中，假設(shè)我們想構(gòu)造一個(gè)包括不同合作者的屬性圖。頂點(diǎn)屬性可能包含用戶名和職業(yè)。我們可以用描述合作者之間關(guān)系的字符串標(biāo)注邊緣。

所得的圖形將具有類型簽名

val userGraph: Graph[(String, String), String]

有很多方式從一個(gè)原始文件、RDD構(gòu)造一個(gè)屬性圖。最一般的方法是利用Graph object。下面的代碼從RDD集合生成屬性圖。

// Assume the SparkContext has already been constructed
val sc: SparkContext
// Create an RDD for the vertices
val users: RDD[(VertexId, (String, String))] =
  sc.parallelize(Array((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),
                       (5L, ("franklin", "prof")), (2L, ("istoica", "prof"))))
// Create an RDD for edges
val relationships: RDD[Edge[String]] =
  sc.parallelize(Array(Edge(3L, 7L, "collab"),    Edge(5L, 3L, "advisor"),
                       Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi")))
// Define a default user in case there are relationship with missing user
val defaultUser = ("John Doe", "Missing")
// Build the initial Graph
val graph = Graph(users, relationships, defaultUser)

在上面的例子中，我們用到了Edge樣本類。邊有一個(gè)srcId和dstId分別對應(yīng)于源和目標(biāo)頂點(diǎn)的標(biāo)示符。另外，Edge類有一個(gè)attr成員用來存儲邊屬性。

我們可以分別用graph.vertices和graph.edges成員將一個(gè)圖解構(gòu)為相應(yīng)的頂點(diǎn)和邊。

val graph: Graph[(String, String), String] // Constructed from above
// Count all users which are postdocs
graph.vertices.filter { case (id, (name, pos)) => pos == "postdoc" }.count
// Count all the edges where src > dst
graph.edges.filter(e => e.srcId > e.dstId).count

注意，graph.vertices返回一個(gè)VertexRDD[(String, String)]，它繼承于 RDD[(VertexID, (String, String))]。所以我們可以用scala的case表達(dá)式解構(gòu)這個(gè)元組。另一方面，
graph.edges返回一個(gè)包含Edge[String]對象的EdgeRDD。我們也可以用到case類的類型構(gòu)造器，如下例所示。

graph.edges.filter { case Edge(src, dst, prop) => src > dst }.count

除了屬性圖的頂點(diǎn)和邊視圖，GraphX也包含了一個(gè)三元組視圖，三元視圖邏輯上將頂點(diǎn)和邊的屬性保存為一個(gè)RDD[EdgeTriplet[VD, ED]]，它包含EdgeTriplet類的實(shí)例。可以通過下面的Sql表達(dá)式表示這個(gè)連接。

SELECT src.id, dst.id, src.attr, e.attr, dst.attr
FROM edges AS e LEFT JOIN vertices AS src, vertices AS dst
ON e.srcId = src.Id AND e.dstId = dst.Id

或者通過下面的圖來表示。

EdgeTriplet類繼承于Edge類，并且加入了srcAttr和dstAttr成員，這兩個(gè)成員分別包含源和目的的屬性。我們可以用一個(gè)三元組視圖渲染字符串集合用來描述用戶之間的關(guān)系。

val graph: Graph[(String, String), String] // Constructed from above
// Use the triplets view to create an RDD of facts.
val facts: RDD[String] =
  graph.triplets.map(triplet =>
    triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1)
facts.collect.foreach(println(_))