Spark Graphx求任意两点最短路径（单机版）

测试数据为1000行点和边数据

vertices_2（起点 终点）（部分数据）：

edges_2（起点 终点 距离）（部分数据）:

代码：

//Pregel算法求单源最短路径
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx._
import org.apache.spark.graphx.{EdgeDirection, Graph, VertexId}
object test {
  //屏蔽不必要的日志显示在终端上
  Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
  Logger.getLogger("org.apache.eclipse.jetty.server").setLevel(Level.OFF)
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("this.getClass().getSimpleName().filter(!_.equals('$'))").setMaster("local")
    val sc = new SparkContext(conf)
    //spark graphx计算最短路径读取edges和vertices文件
    val edgeFile = sc.textFile("D://edges_2.txt")
    val vertexFile = sc.textFile("D://vertices_2.txt")

    //edges
    val edges = edgeFile.map { e =>
      val fields = e.split(" ")
      Edge(fields(0).toLong,fields(1).toLong,fields(2).toDouble)
    }
    //vertex
    val vertices = vertexFile.map{e=>
      val fields = e.split(" ")
      (fields(0).toLong,fields(1))
    }

    val graph = Graph(vertices,edges,"").persist()
    //graph.vertices.foreach(println)
    //graph.edges.foreach(println)

    //val sourceId: VertexId = 1 // The ultimate source

    print("请输入起点ID：")
    val sourceId : VertexId = scala.io.StdIn.readInt()
    println(sourceId)

    //起始时间
    val start = System.currentTimeMillis()

    // Initialize the graph such that all vertices except the root have distance infinity.
    val initialGraph: Graph[(Double, List[VertexId]), Double] = graph.mapVertices((id, _) =>
      if (id == sourceId) (0.0, List[VertexId](sourceId))
      else (Double.PositiveInfinity, List[VertexId]()))

    val sssp = initialGraph.pregel((Double.PositiveInfinity, List[VertexId]()), Int.MaxValue, EdgeDirection.Out)(

      // Vertex Program
      (id, dist, newDist) => if (dist._1 < newDist._1) dist else newDist,

      // Send Message
      triplet => {
        if (triplet.srcAttr._1 < triplet.dstAttr._1 - triplet.attr) {
          Iterator((triplet.dstId, (triplet.srcAttr._1 + triplet.attr, triplet.srcAttr._2 :+ triplet.dstId)))
        } else {
          Iterator.empty
        }
      },
      //Merge Message
      (a, b) => if (a._1 < b._1) a else b)
    //println(sssp.vertices.collect.mkString("\n"))
    //println(sssp.vertices.filter{case(id,v) => id ==3})
    print("请输入终点ID：")
    val end_ID = scala.io.StdIn.readInt()
    println(end_ID)
    println(sssp.vertices.collect.filter{case(id,v) => id == end_ID}.mkString("\n"))

    //程序结束的时间
    val end = System.currentTimeMillis()
    print("查找点"+sourceId+"到点"+end_ID+"花费的时间为："+(end-start)+"毫秒")

    sc.stop()
  }
}

执行效果：