Maximum value in the Pregel formalism in Spark

MaxValue.scala

/**
  * Maximum value algorithm: at each iteration each
  * node sends to its neighbors the largest value it 
  * has ever seen.
  */

package graphalgorithms


import org.apache.spark.graphx._


object MaxValue {


  val MAX_ITERATIONS = 100


  def run(g: Graph[Int, Int]): Graph[Int, Int] = {

    val mergeMsg = (a:Int, b: Int) => math.max(a, b)
    val activeDirection = EdgeDirection.Either
    val numIterations = MAX_ITERATIONS
    val initialMsg = Integer.MIN_VALUE
    val vprog = (vid: VertexId, vd: Int, a: Int) => math.max(vd, a)
    val sendMsg = (etr: EdgeTriplet[Int, Int]) => {
      val a = math.max(etr.srcAttr, etr.dstAttr)
      Iterator((etr.srcId, a), (etr.dstId, a))
    }

    Pregel(g, initialMsg, numIterations)(
      vprog, sendMsg, mergeMsg)
  }


}