GraphX最短路径之实践及心得.docxVIP

GraphX最短路径之实践与心得前言：由于Spark是个新生事物，网上的资料较少，所以作为一个学习者，很高兴能与大家分享一下自己的学习心得，希望对大家有所帮助！有些东西自己也不是很理解，所以这只是一个探讨过程，大家如果有更好的见解和看法，希望大家也通过网络分享出来，这样也可以促进spark的发展，让我们都来为spark的成长做出自己一点小小的贡献吧。另外，再次说一下，Scala语言真的是一门不错的语言。实验内容本实验主要通过构建一张图，然后利用Pregel算法求出从某起始顶点到达其他顶点的最短路径。由于Spark自带的shortpath例子只有求出初始节点到达其他节点最短路径的值，本实验将在原来的基础上求出最短路径的路径图（经过哪些节点）。相关知识Pregel算法思想：Pregel将目标图类问题的运算模型归结为在图的拓扑节点(Vertex)上迭代执行特定的算法。每次迭代称为一个superstep，在Pregel中，数据模型的主要概念包括节点，边（Edge）和消息（Message）。在每个superstep步骤中，各个节点执行相同的用户定义函数来处理数据，更新自身的状态乃至更改整个图的拓扑结构（如增减节点，边等）。每个节点的边则用来链接相关的目标节点，通过发送消息给其它节点来传递数据。整个处理流程中，数据的接收和处理是以superstep为节拍来同步的，在一个superstep中各个节点所发送的消息，直到下一个superstep里才会被目标节点所接收和处理并触发状态变更。这种基于节拍的处理流程，很大程度上简化了数据同步的处理。?每个节点在当前superstep中处理完数据后，会投票决定自身是否停止处理，如果没有被消息再次触发，在以后的superstep中就不会调度该节点进行运算。当所有节点都停止后，整个迭代过程结束。这听起来是不是很抽象，下面就让我们通过代码来看看一个具体的例子：寻找最短路径。总体思路找出度为1的节点代码说明{varg = graph.mapVertices((vid, vdata) = vprog(vid, vdata, initialMsg)).cache()// compute the messagesvarmessages = g.mapReduceTriplets(sendMsg, mergeMsg)varresult : VertexRDD[Array[Edge[Double]]] = messagesvarresults : ArrayBuffer[(VertexId,(Array[Edge[Double]]))] = new ArrayBuffer[(VertexId,(Array[Edge[Double]]))]()varactiveMessages = messages.count()// Loopvari = 0result.collect.map(r = results.+=(r))while (activeMessages 0 i maxIterations) {// Receive the messages. Vertices that didnt get any messages do not appear in newVerts.valnewVerts = g.vertices.innerJoin(messages)(vprog).cache()// Update the graph with the new vertices.g = g.outerJoinVertices(newVerts) { (vid, old, newOpt) = newOpt.getOrElse(old) }g.cache()messages = g.mapReduceTriplets(sendMsg, mergeMsg, Some((newVerts, activeDirection))).cache()activeMessages = messages.count()if(activeMessages 0){// update new message to resultsmessages.collect.foreach{ ve =valvId = ve._1result = result.filter(v = v._1 != vId) results = results.filter(r = r._1 != vId)valedges = ve._2.array vararr1 : ArrayBuffer[Edge[Double]] = new ArrayBuffer[Edge[Double]]()for(es - edges ){varsrcId = es.srcIdvarr1