8

我在 spark 站点上运行了单源最短路径 (SSSP) 示例，如下所示：

graphx-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)

但我需要如下实际路径：
=>
0 -> 0,0
0 -> 2,1
0 -> 3,1
0 -> 2 -> 4,2
0 -> 3 -> 1,2

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

4

2 回答 2

8

您必须修改算法以便不仅存储最短路径长度而且存储实际路径。因此，您应该存储元组，而不是存储Double为顶点的属性：(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"))
  }
}
于 2014-11-24T16:10:44.343 回答
0

可能是过时的答案，但看看这个解决方案Find all paths in graph using Apache Spark

于 2015-08-28T12:59:02.860 回答