pregel是谷歌提出的图计算的一个算法的概念。用于迭代的计算每个顶点的属性，直到满足某个条件（达到稳定状态）。
pregel主要的用途是图遍历（BFS），单源最短路径（Single Source Shortest path）等问题。
GraphX的pregel操作符是一个批量同步的并行消息抽象。pregel执行一系列的super step，在这些步骤中顶点接收到所有的前一个super step的inbound message，然后计算出顶点属性的心值，在将信息发送到下一个super step中。
API的使用格式：
def pregel[A:ClassTag](initialMsg:A, maxIterations:Int = Int.MaxValue, activeDirection:EdgeDirection=EdgeDirection.Either)(vprog:(VertexId, VD, A) => VD, sendMsg:EdgeTriplet[VD, ED] => Iterator[(VertexId, A)], mergeMsg:(A, A) => A): Graph[VD, ED]
在pregel算法中，initialMsg，也就是迭代中传送的参数的类型（pregel message type）应该跟顶点的类型有关系，在vprog中，VD和A应该可以一起计算出VD。
API中的参数：
1. initialMsg：第一次迭代的时候每个顶点获取的初始值。
2. maxIterations：最大迭代的次数
3. activeDirection：在pregel函数中，该参数的默认值设置为EdgeDirection.Either是有原因的。在这里要注意active。API中的解释是：The direction of edges incident to a vertex that received a message in the previous round on which to run “sendMsg”. 该参数的含义是：如果在前一个回合中某个顶点接收到了信息，只有该顶点的某一种边（也就是这里定义的，可能是入边，可能是出边，或者两个都是（也就是Either））在当前回合的迭代中会继续执行sendMsg。所以，至此才能理解pregel的active代表的含义：如果某个顶点收到了信息，该顶点就处于激活状态。处于激活状态的顶点代表状态发生了一些变化，此时他会继续向外“扩张”，也就是调用sendMsg来对周边的其他的点进行改变。这样，其他接收到信息的点也会被激活。逐渐的，没收到信息的点被关闭激活态，直到所有的顶点都处于非激活态，则pregel算法结束。
所以，pregel算法与pagerank不同，他不提供参数用于执行到满足一定条件时（pagerank可以设置在差异小于一定值之后就停止执行）直接停止执行，要么执行满足一定次数，要么一直等到所有顶点都关闭。所以即使maxIterations设置的很大，也有可能很快就结束了。
所以，pregel算法类似于一个循环，在这个循环中，必须有打破循环一直执行的变量的改变，这个变量的改变就是sendMsg中发送到某个顶点的信息。该方法应该只在适当的条件下对邻居节点产生影响。而不应该一直变动。否则这个图是不会消停的。
4. vprog：用户定义的vertex program，在每个收到inbound message的顶点上进行并行计算（也就是该顶点的计算结果不受其他顶点的计算结果的影响），计算出顶点的一个新的值。
vprog在实际算法中的使用：
var g = mapVertices((vid, vdata) => vprog(vid, vdata, initialMsg) ).cache()//顶点接收initialMsg，然后调用vprog生成一个新的值。生成的新的值的类型跟原来的图应该是一样的。
在迭代计算的过程里，第一次的计算中，每个顶点都会收到initialMsg并参与计算。此后，只有收到消息的顶点才会参与计算。
要注意vdata和initMsg。他们之间没有什么关联。vdata是图的顶点属性，initMsg则是pregel初始值要跟vdata一起，在vprog中进行计算，这个值才是计算开始时，图的顶点属性的值。
5. sendMsg：顶点往外发送信息。注意，发送出的信息类型是Iterator[(VertexId, A)]，也就是这里发送信息并不是每个合适的边都要发送信息。只是一部分。（这样才能实现发送的信息越来越少，最后停止执行）
6. mergeMsg：将收到的信息进行合并。
要注意区别mergeMsg和vprog的区别。mergeMsg仅仅是针对顶点收到的信息进行处理，而vprog是对收到的信息与顶点原有属性进行计算。他们常常很像是原因是顶点的原有属性跟收到的信息的数据类型是一样的。但要明白他们的作用的区别。
关于pregel的paper：
https://kowshik.github.io/JPregel/pregel_paper.pdf
如何使用pregel实现计算单源最短路径：
1. 什么是单源最短路径：
给定一个带权有向图，其中每个边的权都是一个实数。此外，给定V中的一个顶点，称为源。要计算从源到其他各个顶点的最短路径长度。这个长度就是指路上各个边的权的和。这个问题称为单源最短路径问题。
计算单源最短路径的方法：
（1）Dijkstra算法：
最短路径的最优子结构性质：如果P(i, j)={Vi, ..Vk, ..Vs,….Vj}是从i到j的最短路径，k和s是这条路径中的中间顶点，那么P(k, s)必定是从k到s的最短路径。
迪科斯彻算法应用了贪心算法思想。贪心算法是指：再对问题求解的时候，总是做当前看来是最好的选择。也就是不从整体最优上加以考虑，但这却往往能得到最优解。贪心策略的选择必须具备无后效性，也就是整个状态中以前的过程不会影响以后的状态，只跟当前状态有关。
使用pregel来实现Dijkstra的思路是：从原点出发，依次计算其每一层的子节点到原点的最短距离，直到迭代到了所有的连通的顶点。此时就可以计算出每个顶点跟原点之间的最短距离。
spark官网给出的pregel算法是：
val sourceId = 12;//指定原点
val initGraph = graph.mapVertices((id, _) =>if(id == sourceId) 0 else Double.maxValue)//将图中的所有的点（除了原点）都设置属性为无限大（代表当前这些所有的点距离原点的距离都是无限远）
val maxIter = Int.maxValue
val direction = EdgeDirection.Either
val initValue = Double.maxValue//第一次迭代时，每个顶点获取的值是最大值
val vprog = (id:VertexId, dist:Double, newDist:Double) => math.min(dist, newDist)//顶点对所获取的相邻顶点发送的信息进行处理后的结果（newDist）进行处理。跟当前顶点已经记录的距离原点的距离进行比较，保留较小的值。
val mergeMsg = (a:Double, b:Double) => math.min(a, b)//每个顶点收集到的临界顶点发送过来消息汇总，取出发来的最小值。
val sendMsg = (triplet:EdgeTriplet[Double, Double]) => {
if(triplet.srcAttr+triplet.attr < triplet.dstAttr){
Iterator((triplet.dstAttr, triplet.srcAttr+triplet.attr))
}else{
Iterator.Empty
}
}//一个triplet里，如果src中存储的到原点的距离加上当前边权小于dst顶点到原点的距离，就说明dst中保存的到原点的距离并不是最小的。就需要向dst顶点发送消息。
val result = initGraph.pregel(initValue, maxIter, direction)(vprog, sendMsg, mergeMsg)
在这种方法里有个问题：只计算出了最短路径，但没有保留这个路径。要保留路径，就需要将图的顶点的数据结构进行修改，使其在计算最短路径时也会保留最短路径。
import org.apache.spark.graphx._
import org.apache.spark.graphx.util.GraphGenerators
import org.apache.spark.sql.SparkSession
import com.proud.ark.config.ConfigUtil
val graph:Graph[Long, Double] = GraphGenerators.logNormalGraph(sc, 100).mapEdges(x => x.attr.toDouble)
val sourceId:VertexId = 42;
val initGraph:Graph[(Double, List[VertexId]), Double] = graph.mapVertices((id, _) => if(id == sourceId) (0.0, List.empty) else (Double.MaxValue, List.empty))
val initMsg:(Double, List[VertexId]) = (Double.MaxValue, List.empty)
val maxIter = Int.MaxValue
val edgeDirection = EdgeDirection.In
val vprog = (id:VertexId, a:(Double, List[VertexId]), b:(Double, List[VertexId])) => if(a._1 < b._1){
a
}else{
b
}
val sendMsg:(EdgeTriplet[(Double, List[VertexId]), Double] => Iterator[(VertexId, (Double, List[VertexId]))]) = (triplet:EdgeTriplet[(Double, List[VertexId]), Double]) => {
if(triplet.srcAttr._1+triplet.attr < triplet.dstAttr._1){
Iterator((triplet.dstId, (triplet.srcAttr._1+triplet.attr, triplet.dstId+:triplet.srcAttr._2)))
}else{
Iterator.empty
}
}
val mergeMsg:(((Double, List[VertexId]), (Double, List[VertexId])) => (Double, List[VertexId])) = (a:(Double, List[VertexId]), b:(Double, List[VertexId])) => {
if(a._1 < b._1){
a
}else{
b
}
}
val sssp = initGraph.pregel(initMsg, maxIter, edgeDirection)(vprog, sendMsg, mergeMsg)
思考题：如果要保存若干条最短路径该如何解决？
