• Pregel API

    Pregel API

    图本身是递归数据结构，顶点的属性依赖于它们邻居的属性，这些邻居的属性又依赖于自己邻居的属性。所以许多重要的图算法都是迭代的重新计算每个顶点的属性，直到满足某个确定的条件。
    一系列的graph-parallel抽象已经被提出来用来表达这些迭代算法。GraphX公开了一个类似Pregel的操作，它是广泛使用的Pregel和GraphLab抽象的一个融合。

    在GraphX中，更高级的Pregel操作是一个约束到图拓扑的批量同步（bulk-synchronous）并行消息抽象。Pregel操作者执行一系列的超级步骤（super steps），在这些步骤中，顶点从
    之前的超级步骤中接收进入(inbound)消息的总和，为顶点属性计算一个新的值，然后在以后的超级步骤中发送消息到邻居顶点。不像Pregel而更像GraphLab，消息作为一个边三元组的函数被并行
    计算，消息计算既访问了源顶点特征也访问了目的顶点特征。在超级步中，没有收到消息的顶点被跳过。当没有消息遗留时，Pregel操作停止迭代并返回最终的图。

    注意，与更标准的Pregel实现不同的是，GraphX中的顶点仅仅能发送信息给邻居顶点，并利用用户自定义的消息函数构造消息。这些限制允许在GraphX进行额外的优化。

    一下是@pregel[A](A,Int,EdgeDirection"> Pregel操作((VertexId,VD,A)⇒VD,(EdgeTriplet[VD,ED])⇒Iterator[(VertexId,A)],(A,A)⇒A)(ClassTag[A]):Graph[VD,ED])的类型签名以及实现草图（注意，访问graph.cache已经被删除）

    1. class GraphOps[VD, ED] {
    2.   def pregel[A]
    3.       (initialMsg: A,
    4.        maxIter: Int = Int.MaxValue,
    5.        activeDir: EdgeDirection = EdgeDirection.Out)
    6.       (vprog: (VertexId, VD, A) => VD,
    7.        sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexId, A)],
    8.        mergeMsg: (A, A) => A)
    9.     : Graph[VD, ED] = {
    10.     // Receive the initial message at each vertex
    11.     var g = mapVertices( (vid, vdata) => vprog(vid, vdata, initialMsg) ).cache()
    12.     // compute the messages
    13.     var messages = g.mapReduceTriplets(sendMsg, mergeMsg)
    14.     var activeMessages = messages.count()
    15.     // Loop until no messages remain or maxIterations is achieved
    16.     var i = 0
    17.     while (activeMessages > 0 && i < maxIterations) {
    18.       // Receive the messages: -----------------------------------------------------------------------
    19.       // Run the vertex program on all vertices that receive messages
    20.       val newVerts = g.vertices.innerJoin(messages)(vprog).cache()
    21.       // Merge the new vertex values back into the graph
    22.       g = g.outerJoinVertices(newVerts) { (vid, old, newOpt) => newOpt.getOrElse(old) }.cache()
    23.       // Send Messages: ------------------------------------------------------------------------------
    24.       // Vertices that didn't receive a message above don't appear in newVerts and therefore don't
    25.       // get to send messages.  More precisely the map phase of mapReduceTriplets is only invoked
    26.       // on edges in the activeDir of vertices in newVerts
    27.       messages = g.mapReduceTriplets(sendMsg, mergeMsg, Some((newVerts, activeDir))).cache()
    28.       activeMessages = messages.count()
    29.       i += 1
    30.     }
    31.     g
    32.   }
    33. }

    注意，pregel有两个参数列表（graph.pregel(list1)(list2)）。第一个参数列表包含配置参数初始消息、最大迭代数、发送消息的边的方向（默认是沿边方向出）。第二个参数列表包含用户
    自定义的函数用来接收消息（vprog）、计算消息（sendMsg）、合并消息（mergeMsg）。

    我们可以用Pregel操作表达计算单源最短路径( single source shortest path)。

    1. import org.apache.spark.graphx._
    2. // Import random graph generation library
    3. import org.apache.spark.graphx.util.GraphGenerators
    4. // A graph with edge attributes containing distances
    5. val graph: Graph[Int, Double] =
    6.   GraphGenerators.logNormalGraph(sc, numVertices = 100).mapEdges(e => e.attr.toDouble)
    7. val sourceId: VertexId = 42 // The ultimate source
    8. // Initialize the graph such that all vertices except the root have distance infinity.
    9. val initialGraph = graph.mapVertices((id, _) => if (id == sourceId) 0.0 else Double.PositiveInfinity)
    10. val sssp = initialGraph.pregel(Double.PositiveInfinity)(
    11.   (id, dist, newDist) => math.min(dist, newDist), // Vertex Program
    12.   triplet => {  // Send Message
    13.     if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
    14.       Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
    15.     } else {
    16.       Iterator.empty
    17.     }
    18.   },
    19.   (a,b) => math.min(a,b) // Merge Message
    20.   )
    21. println(sssp.vertices.collect.mkString("\n"))