Pregel API

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

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

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

一下是 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"))