Implement PageRank on VertexProgram API

src/main/scala/com/algorithm/PageRank.scala

@@ -0,0 +1,42 @@
+package com.algorithm
+
+// Non-normalized PageRank
+// Assumes no sink vertices
+class PageRank(iters: Int) extends VertexProgram[Int, Double, Double, Double] {
+
+ require(iters >= 1)
+
+ override val mode: VertexProgram.Mode = VertexProgram.Outwards
+
+ override def gather(edgeVal: Int, message: Double): Double = message
+
+ override def sum(a: Double, b: Double): Double = a + b
+
+ override def apply(superStepNumber: Int, thisVertex: VertexInfo, oldVal: Double, total: Option[Double]): Double = {
+ if (superStepNumber == 0) {
+ // Do nothing on first superstep except send PR to neighbours
+ oldVal
+ } else {
+ // println(s"StepNum: ${superStepNumber}")
+ val sum = total.getOrElse(0.0) // Safe: should have received something from every neighbour
+ 0.15 + (0.85 * sum)
+ }
+ }
+
+ override def scatter(superStepNumber: Int, thisVertex: VertexInfo, oldVal: Double, newVal: Double): Option[Double] = {
+ if(superStepNumber >= iters) {
+ None
+ } else {
+ Some(newVal / thisVertex.degree)
+ }
+ }
+
+ override def voteToHalt(superStepNumber: Int, oldVal: Double, newVal: Double): Boolean = {
+ superStepNumber >= iters
+ }
+
+ override val defaultVertexValue: Double = 1.0
+
+ override val defaultActivationStatus: Boolean = true
+
+}

src/test/scala/com/algorithm/EpsilonCloseMatcher.scala

@@ -0,0 +1,32 @@
+package com.algorithm
+
+import org.scalatest.matchers.MatchResult
+import org.scalatest.matchers.Matcher
+
+object EpsilonCloseMatching {
+
+ class EpsilonCloseCheck[T](epsilon: Double, expected: Map[T, Double]) extends Matcher[Map[T, Double]] {
+
+ def apply(left: Map[T, Double]): MatchResult = {
+
+ val bad = expected.filter { case (k, v) => {
+ left.get(k) match {
+ case None => false
+ case Some(v_prime) => Math.abs(v - v_prime) > epsilon
+ }
+ }}
+
+ MatchResult(
+ left.keySet == expected.keySet && bad.isEmpty,
+ s"Map ${left} not sufficiently close to Map ${expected}",
+ "Maps sufficiently close"
+ )
+ }
+
+ }
+
+ def BeEpsilonClose[T](epsilon: Double)(expected: Map[T, Double]): Matcher[Map[T, Double]] = new EpsilonCloseCheck(epsilon, expected)
+
+}
+
+

src/test/scala/com/algorithm/SequentialTest.scala

@@ -1,6 +1,5 @@
 package com.algorithm
 
-import com.algorithm._
 import org.scalatest._
 import scalax.collection.Graph
 import scalax.collection.GraphPredef._
@@ -211,5 +210,58 @@ class SequentialTest extends FunSuite with Matchers {
  results should be (finalComponents)
  }
 
+ test("PageRank 1") {
+ val g: Graph[Int, WDiEdge] = Graph(
+ 0 ~> 1 % 1,
+ 0 ~> 2 % 1,
+ 1 ~> 2 % 1,
+ 2 ~> 0 % 1,
+ 3 ~> 2 % 1,
+ )
+ val results = SequentialRun(new PageRank(40), g)
+ val normalized = for((vtx, rank) <- results) yield (vtx -> rank/g.nodes.size)
+ val expected = ListMap(
+ g.Node(0) -> 0.372,
+ g.Node(1) -> 0.196,
+ g.Node(2) -> 0.394,
+ g.Node(3) -> 0.037
+ )
+ // object EpsilonMatcher extends EpsilonCloseMatchers(0.01)
+ normalized should EpsilonCloseMatching.BeEpsilonClose[g.NodeT](0.01)(expected)
+ }
+
+ test("PageRank 2") {
+ val g: Graph[Int, WDiEdge] = Graph(
+ 0 ~> 1 % 1,
+ 0 ~> 2 % 1,
+ 0 ~> 3 % 1,
+ 1 ~> 0 % 1,
+ 2 ~> 0 % 1,
+ 3 ~> 0 % 1,
+ 3 ~> 4 % 1,
+ 3 ~> 5 % 1,
+ 3 ~> 6 % 1,
+ 4 ~> 0 % 1,
+ 5 ~> 0 % 1,
+ 5 ~> 2 % 1,
+ 6 ~> 3 % 1,
+ 7 ~> 5 % 1,
+ 7 ~> 6 % 1,
+ )
+ val results = SequentialRun(new PageRank(40), g)
+ val normalized = for((vtx, rank) <- results) yield (vtx -> rank/g.nodes.size)
+ val expected = ListMap(
+ g.Node(0) -> 0.358,
+ g.Node(1) -> 0.120,
+ g.Node(2) -> 0.147,
+ g.Node(3) -> 0.174,
+ g.Node(4) -> 0.056,
+ g.Node(5) -> 0.064,
+ g.Node(6) -> 0.064,
+ g.Node(7) -> 0.019
+ )
+ normalized should EpsilonCloseMatching.BeEpsilonClose[g.NodeT](0.001)(expected)
+ }
+
 
 }