使用Apache Spark Graphx的广度优先搜索算法

Question

我正在尝试使用Apache Spark Graphx实现BFS(广度优先搜索)算法。

这是我目前的实现:

object BFSAlgorithm {

  def run(graph: Graph[VertexId, Int], sourceVertex: VertexId): Graph[Int, Int] = {

    val bfsGraph: Graph[Int, Int] = graph.mapVertices((vertex, _) =>
      if (vertex == sourceVertex) {
        0
      } else {
        Int.MaxValue
      }
    )

    var queue: Queue[VertexId] = Queue[VertexId](sourceVertex)
    while(queue.nonEmpty){
      val currentVertexId = queue.dequeue()
      val neighbours: RDD[EdgeTriplet[Int, Int]] = bfsGraph.triplets.filter(_.srcId == currentVertexId)
      for(triplet <- neighbours){
        if(triplet.dstAttr == Int.MaxValue){
          queue += triplet.dstId
        }
        val distance = triplet.srcAttr + 1
        if(distance < triplet.dstAttr){
          // Update vertex attibute
          bfsGraph.mapVertices((vertex, _) => if(vertex == triplet.dstId) distance else triplet.dstAttr)
        }
      }
    }
    bfsGraph
  }

}

当我尝试在线更新顶点属性时，我收到空指针异常：

bfsGraph.mapVertices((vertex, _) => if(vertex == triplet.dstId) distance else triplet.dstAttr)

我很困惑，因为for循环bfsGraph.vertices为空。

有人能解释我为什么吗？更新图形中顶点属性的最佳方法是什么？

Answer 1

此实现不起作用，因为您正试图访问另一个RDD内的RDD。当您在邻居上调用for循环时，GraphX试图为循环收集一个闭包，该闭包包含循环体中所需变量的列表，在这种情况下，它涉及另一个RDD (bsfGraph)，导致< code > NullPointerException 。