Find the fringe set for each vertex using GraphX

fringe-set.scala

// Depends on AllPairsShortestPaths: https://github.com/apache/spark/pull/3619

import org.apache.spark.graphx._
import org.apache.spark.graphx.lib._

val edges = sc.parallelize((0 until 10).map(x => Edge(x, x + 1, 1)))
val graph = Graph.fromEdges(edges, 1)

val dists = AllPairsShortestPaths.run(graph).cache()
val maxDists = dists.mapValues(_._2).reduceByKey((a, b) => if (a > b) a else b)
val fringes = dists.cogroup(maxDists).mapValues {
  case (candidates, maxDistSeq) =>
    val maxDist = maxDistSeq.head
    candidates.filter(_._2 == maxDist).map(_._1)
}

fringes.foreach(println)
// (0,List(10))
// (1,List(10))
// (2,List(10))
// (3,List(10))
// (4,List(10))
// (5,List(0, 10))
// (6,List(0))
// (7,List(0))
// (8,List(0))
// (9,List(0))
// (10,List(0))