广度优先遍历

  1. val graph = GraphLoader.edgeListFile(sc, "graphx/data/test_graph.txt")
  3. val root: VertexId = 1
  4. val initialGraph = graph.mapVertices((id, _) => if (id == root) 0.0 else
  5. Double.PositiveInfinity)
  7. val vprog = { (id: VertexId, attr: Double, msg: Double) => math.min(attr,msg) }
  9. val sendMessage = { (triplet: EdgeTriplet[Double, Int]) =>
  10.         var iter:Iterator[(VertexId, Double)] = Iterator.empty
  11.         val isSrcMarked = triplet.srcAttr != Double.PositiveInfinity
  12.         val isDstMarked = triplet.dstAttr != Double.PositiveInfinity
  13.         if(!(isSrcMarked && isDstMarked)){
  14.                if(isSrcMarked){
  15.                 iter = Iterator((triplet.dstId,triplet.srcAttr+1))
  16.               }else{
  17.                 iter = Iterator((triplet.srcId,triplet.dstAttr+1))
  18.                }
  19.         }
  20.         iter
  21. }
  23. val reduceMessage = { (a: Double, b: Double) => math.min(a,b) }
  25. val bfs = initialGraph.pregel(Double.PositiveInfinity, 20)(vprog, sendMessage, reduceMessage)
  27. println(bfs.vertices.collect.mkString("\n"))