如何使用Spark GraphX打印最短路径的具体节点与边详情而非仅长度

核心修改逻辑：将原有顶点存储的单一距离属性，扩展为(最短距离: Double, 路径列表: List[VertexId])的二元组，在Pregel迭代过程中同步更新路径信息。

完整可运行代码 #

import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object GraphxShortestPathWithRoute {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 修正原代码中顶点数组缺左括号的语法错误
    val vertexArray = Array(
      (1L, ("Alice1", 28)),
      (2L, ("Bob2", 27)),
      (3L, ("Charlie3", 65)),
      (4L, ("David4", 42)),
      (5L, ("Ed5", 55)),
      (6L, ("Fran6", 50))
    )
    val edgeArray = Array(
      Edge(2L, 1L, 1),
      Edge(2L, 4L, 1),
      Edge(3L, 2L, 1),
      Edge(3L, 6L, 1),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 1),
      Edge(5L, 3L, 1),
      Edge(5L, 6L, 1)
    )

    val vertexRDD = sc.parallelize(vertexArray)
    val edgeRDD = sc.parallelize(edgeArray)
    val graph = Graph(vertexRDD, edgeRDD)

    val sourceId: VertexId = 5L
    // 初始化顶点属性：(距离, 路径列表)，源点路径初始为自身ID，其他为空
    val initialGraph = graph.mapVertices((id, _) =>
      if (id == sourceId) (0.0, List[VertexId](sourceId))
      else (Double.PositiveInfinity, List[VertexId]())
    )

    val sssp = initialGraph.pregel((Double.PositiveInfinity, List[VertexId]()))(
      // 顶点接收消息：取距离更小的属性，更新距离和路径
      (id, attr, newAttr) => if (newAttr._1 < attr._1) newAttr else attr,
      // 发送消息：如果通过当前边到目标点距离更短，发送新的距离和拼接后的路径
      triplet => {
        if (triplet.srcAttr._1 + triplet.attr < triplet.dstAttr._1) {
          Iterator((triplet.dstId, (triplet.srcAttr._1 + triplet.attr, triplet.srcAttr._2 :+ triplet.dstId)))
        } else {
          Iterator.empty
        }
      },
      // 合并消息：取距离更小的消息
      (a, b) => if (a._1 < b._1) a else b
    )

    // 格式化打印结果
    sssp.vertices.collect.foreach { case (vid, (dist, path)) =>
      println(s"顶点$sourceId 到顶点$vid 最短距离为$dist，路径为：${path.mkString("->")}")
    }

    sc.stop()
  }
}

运行输出样例 #

顶点5 到顶点4 最短距离为2.0，路径为：5->2->4
顶点5 到顶点1 最短距离为2.0，路径为：5->2->1
顶点5 到顶点6 最短距离为1.0，路径为：5->6
顶点5 到顶点3 最短距离为1.0，路径为：5->3
顶点5 到顶点5 最短距离为0.0，路径为：5
顶点5 到顶点2 最短距离为1.0，路径为：5->2

内容的提问来源于stack exchange，提问作者Jack