距离问题详解
    image.png
    边的属性代表距离，箭头代表方向，如只能从5到2，不能从2到5，计算顶点5到各点的距离

    1. // 顶点5到其他各顶点的最短距离。聚合操作(Pregel API)
    3. // 消息按照我的理解，就是A点如果更新了，就会向在其他机子上的A点传递自己的信息
    4. val sourceId: VertexId = 5L
    5. //初始化，5到5的距离是零，5到其各个点的距离是无穷
    6. val initailGraph: Graph[Double, Int] = graph.mapVertices((id, _) => if (id == sourceId) 0.0 else Double.PositiveInfinity)
    8. //pregerl api
    9. /*
    10.  def pregel[A: ClassTag](
    11.  //初始值，或者看成初始消息
    12.       initialMsg: A,
    13.       maxIterations: Int = Int.MaxValue,
    14.       //边的方向，选择out，则只计算出边，either 出边或入边
    15.       activeDirection: EdgeDirection = EdgeDirection.out)(
    16.       // 顶点接收到消息后的计算，第一次所有点都调用，传入初始消息，后续只有接收到消息的点会调用
    17.       vprog: (VertexId, VD, A) => VD,
    18.       // 只有在当前迭代中接收到消息的点会调用，由于第一次所有点都有消息，所以所有点都会调用，调用时会沿出边调用triplet
    19.       sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexId, A)],
    20.       // 合并消息，如果一个点有多个消息，先合并
    21.       mergeMsg: (A, A) => A)
    22.     : Graph[VD, ED] = {
    23.     Pregel(graph, initialMsg, maxIterations, activeDirection)(vprog, sendMsg, mergeMsg)
    24.   }
    25.   */
    26. val disGraph: Graph[Double, Int] = initailGraph.pregel(Double.PositiveInfinity, maxIterations = Int.MaxValue,activeDirection = EdgeDirection.Out)(
    27.   // 两个消息来的时候，取其中的最小路径
    28.   (id, dist, newDist) => math.min(dist, newDist),
    30.   // Send Message 函数
    31.   triplet => {
    32.     println(triplet)
    33.     if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
    34.       Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
    35.     } else
    36.     Iterator.empty
    37.   },
    39.   // mergeMsg
    40.   (dista, distb) => math.min(dista, distb)
    41. )
    43. disGraph.vertices.foreach(println)

    可以比较一下EdgeDirection = EdgeDirection.Out和EdgeDirection = EdgeDirection.Either的打印结果
    当点更新后，就会向另一个地方的自己发送消息，从而触发triplet的迭代，
    最终结果一致，但out的迭代次数更少
    image.png