如何通过apache火花图形获取SSSP实际路径？

Question

我在火花站点上运行了单源最短路径（SSSP）示例，如下所示：

Grax-SSSP预凝胶实例

代码（scala）：

object Pregel_SSSP {
def main(args: Array[String]) {
val sc = new SparkContext("local", "Allen Pregel Test", System.getenv("SPARK_HOME"), SparkContext.jarOfClass(this.getClass))
// A graph with edge attributes containing distances
val graph: Graph[Int, Double] =
  GraphGenerators.logNormalGraph(sc, numVertices = 5).mapEdges(e => e.attr.toDouble)
graph.edges.foreach(println)
val sourceId: VertexId = 0 // The ultimate source

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

val sssp = initialGraph.pregel(Double.PositiveInfinity, Int.MaxValue, EdgeDirection.Out)(

  // Vertex Program
  (id, dist, newDist) => math.min(dist, newDist),

  // Send Message
  triplet => {
    if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
      Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
    } else {
      Iterator.empty
    }
  },
  //Merge Message
  (a, b) => math.min(a, b))
println(sssp.vertices.collect.mkString("\n"))
 } 
}

sourceId:0获取结果：
（0,0.0）
（4,2.0）
（2,1.0）
（3,1.0）
（1,2.0）

但是我需要如下实际路径：
=

如何通过spark graphX获得SSSP实际路径<有人给我一些提示吗<谢谢你的帮助

Answer 1

也许这是过时的答案，但看看这个解决方案，使用ApacheSpark查找图形中的所有路径

Answer 2

为了存储最短路径长度和实际路径，必须修改算法。因此，不要将Double存储为顶点的属性，而应该存储tuple:（Double，List[VertexId]）也许这个代码对您有用。

object Pregel_SSSP {
  def main(args: Array[String]) {
    val sc = new SparkContext("local", "Allen Pregel Test", System.getenv("SPARK_HOME"), SparkContext.jarOfClass(this.getClass))
    // A graph with edge attributes containing distances
    val graph: Graph[Int, Double] =
      GraphGenerators.logNormalGraph(sc, numVertices = 5).mapEdges(e => e.attr.toDouble)
    graph.edges.foreach(println)
    val sourceId: VertexId = 0 // The ultimate source

    // 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"))
  }
}