Spark GraphX圖算法

GraphX包括一組圖算法來(lái)簡(jiǎn)化分析任務(wù)。這些算法包含在org.apache.spark.graphx.lib包中，可以被直接訪問(wèn)。

PageRank算法

PageRank度量一個(gè)圖中每個(gè)頂點(diǎn)的重要程度，假定從u到v的一條邊代表v的重要性標(biāo)簽。例如，一個(gè)Twitter用戶被許多其它人粉，該用戶排名很高。GraphX帶有靜態(tài)和動(dòng)態(tài)PageRank的實(shí)現(xiàn)方法，這些方法在PageRank object中。靜態(tài)的PageRank運(yùn)行固定次數(shù)的迭代，而動(dòng)態(tài)的PageRank一直運(yùn)行，直到收斂。[GraphOps]()允許直接調(diào)用這些算法作為圖上的方法。

GraphX包含一個(gè)我們可以運(yùn)行PageRank的社交網(wǎng)絡(luò)數(shù)據(jù)集的例子。用戶集在graphx/data/users.txt中，用戶之間的關(guān)系在graphx/data/followers.txt中。我們通過(guò)下面的方法計(jì)算每個(gè)用戶的PageRank。

// Load the edges as a graph
val graph = GraphLoader.edgeListFile(sc, "graphx/data/followers.txt")
// Run PageRank
val ranks = graph.pageRank(0.0001).vertices
// Join the ranks with the usernames
val users = sc.textFile("graphx/data/users.txt").map { line =>
  val fields = line.split(",")
  (fields(0).toLong, fields(1))
}
val ranksByUsername = users.join(ranks).map {
  case (id, (username, rank)) => (username, rank)
}
// Print the result
println(ranksByUsername.collect().mkString("\n"))

連通體算法

連通體算法用id標(biāo)注圖中每個(gè)連通體，將連通體中序號(hào)最小的頂點(diǎn)的id作為連通體的id。例如，在社交網(wǎng)絡(luò)中，連通體可以近似為集群。GraphX在ConnectedComponents object中包含了一個(gè)算法的實(shí)現(xiàn)，我們通過(guò)下面的方法計(jì)算社交網(wǎng)絡(luò)數(shù)據(jù)集中的連通體。

/ Load the graph as in the PageRank example
val graph = GraphLoader.edgeListFile(sc, "graphx/data/followers.txt")
// Find the connected components
val cc = graph.connectedComponents().vertices
// Join the connected components with the usernames
val users = sc.textFile("graphx/data/users.txt").map { line =>
  val fields = line.split(",")
  (fields(0).toLong, fields(1))
}
val ccByUsername = users.join(cc).map {
  case (id, (username, cc)) => (username, cc)
}
// Print the result
println(ccByUsername.collect().mkString("\n"))

三角形計(jì)數(shù)算法

一個(gè)頂點(diǎn)有兩個(gè)相鄰的頂點(diǎn)以及相鄰頂點(diǎn)之間的邊時(shí)，這個(gè)頂點(diǎn)是一個(gè)三角形的一部分。GraphX在TriangleCount object中實(shí)現(xiàn)了一個(gè)三角形計(jì)數(shù)算法，它計(jì)算通過(guò)每個(gè)頂點(diǎn)的三角形的數(shù)量。需要注意的是，在計(jì)算社交網(wǎng)絡(luò)數(shù)據(jù)集的三角形計(jì)數(shù)時(shí)，TriangleCount需要邊的方向是規(guī)范的方向(srcId < dstId),并且圖通過(guò)Graph.partitionBy分片過(guò)。

// Load the edges in canonical order and partition the graph for triangle count
val graph = GraphLoader.edgeListFile(sc, "graphx/data/followers.txt", true).partitionBy(PartitionStrategy.RandomVertexCut)
// Find the triangle count for each vertex
val triCounts = graph.triangleCount().vertices
// Join the triangle counts with the usernames
val users = sc.textFile("graphx/data/users.txt").map { line =>
  val fields = line.split(",")
  (fields(0).toLong, fields(1))
}
val triCountByUsername = users.join(triCounts).map { case (id, (username, tc)) =>
  (username, tc)
}
// Print the result
println(triCountByUsername.collect().mkString("\n"))