Pregel概述

• Pregel是Google提出的用于大規模分布式圖計算框架
  ? - 圖遍歷(BFS)
  ? - 單源最短路徑(SSSP)
  ? - PageRank計算
• Pregel的計算由一系列迭代組成，稱為supersteps
• Pregel迭代過程
  ? - 每個頂點從上一個superstep接收入站消息
  ? - 計算頂點新的屬性值
  ? - 在下一個superstep中想相鄰的頂點發送消息
  ? - 當沒有剩余消息是，迭代結束

Pregel原理分析

pregel函數源碼以及各個參數的簡介：

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(graph, initialMsg, maxIterations, activeDirection)(vprog, sendMsg, mergeMsg)} 參數說明

initialMsg	圖初始化的時候，開始模型計算的時候，所有節點都會收到一個消息
maxIterations	最大迭代次數
activeDirection	規定了發送消息的方向
vprog	節點接收改消息將聚合后的數據和本節點進行屬性的合并
sendMsg	激活狀態節點調用該方法發送消息
mergeMsg	如果一個節點接收到多條消息，先用mergeMsg來將多條消息聚合成為一條消息，如果節點只收到一條消息，則不調用改函數

案例：求5頂點到1頂點的最短距離

• 頂點的兩個狀態：
  ? - 鈍化態：類似于休眠，不做任何處理
  ? - 激活態：可以進行數據的接受和發送
• 頂點能夠處于激活狀態需要的條件
  ? - 成功收到消息
  ? - 成功發送任何一條消息
  
  首先，給5節點的初始值為0，其他值的初始值為正無窮大
  
  然后從5節點出發，每次用發送方的值+權重和接受方的值相比較，取小值作為接受方的值，然后再以此方式發送給下一節點，如果發送方的值+權重大于接受方的值，則無法發送給下一節點。根據這種方法得到的結果如下圖：
  
  最終得到的5頂點到1頂點的最短距離為15。
  代碼實現：

val conf = new SparkConf().setMaster("local[2]").setAppName("mytst")val sc = SparkContext.getOrCreate(conf)val vect = sc.parallelize(Array((1L, ("Alice", 28)),(2L, ("Bob", 27)),(3L, ("Charlie", 65)),(4L, ("David", 42)),(5L, ("Ed", 55)),(6L, ("Fran", 50))))val edges = sc.parallelize(Array(Edge(2L, 1L, 7),Edge(2L, 4L, 2),Edge(3L, 2L, 4),Edge(3L, 6L, 3),Edge(4L, 1L, 1),Edge(2L, 5L, 2),Edge(5L, 3L, 8),Edge(5L, 6L, 3)))val graphx = Graph(vect,edges)// 設置起始頂點val srcVectId = 5Lval initialGraph = graphx.mapVertices({case (vid,(name,age))=>if (vid==5L) 0.0 else Double.PositiveInfinity})// 調用pregelval pregelGraph = initialGraph.pregel(Double.PositiveInfinity, //每個點的初始值，無窮大Int.MaxValue, //最大迭代次數EdgeDirection.Out //發送信息的方向)( //vprog(接受到的消息和自己的消息進行合并)(vid:VertexId,vd:Double,distMsg:Double)=>{val minDist = math.min(vd,distMsg)println(s"頂點${vid}，屬性${vd}，收到消息${distMsg}，合并后的屬性${minDist}")minDist},(edgeTriplet:EdgeTriplet[Double,PartitionID])=>{ //發送消息,如果自己的消息+權重<目的地的消息，則發送if(edgeTriplet.srcAttr+edgeTriplet.attr<edgeTriplet.dstAttr){println(s"頂點${edgeTriplet.srcId} 給頂點${edgeTriplet.dstId} 發送消息 ${edgeTriplet.srcAttr + edgeTriplet.attr}")Iterator[(VertexId,Double)]((edgeTriplet.dstId,edgeTriplet.srcAttr+edgeTriplet.attr))}else{Iterator.empty}},(msg1:Double,msg2:Double)=>math.min(msg1,msg2) //多條接收消息，mergeMessage,取小合并多條消息)// 輸出結果pregelGraph.triplets.foreach(println)println(pregelGraph.vertices.collect.mkString(","))// 關閉資源sc.stop()

迭代結果展示：

//------------------------------------------ 各個頂點接受初始消息initialMsg ------------------------------------------
頂點3，屬性Infinity，收到消息Infinity，合并后的屬性Infinity
頂點2，屬性Infinity，收到消息Infinity，合并后的屬性Infinity
頂點4，屬性Infinity，收到消息Infinity，合并后的屬性Infinity
頂點6，屬性Infinity，收到消息Infinity，合并后的屬性Infinity
頂點1，屬性Infinity，收到消息Infinity，合并后的屬性Infinity
頂點5，屬性0.0，收到消息Infinity，合并后的屬性0.0
//------------------------------------------ 第一次迭代 ------------------------------------------
頂點5 給 頂點6 發送消息 3.0
頂點5 給 頂點3 發送消息 8.0
頂點3，屬性Infinity，收到消息8.0，合并后的屬性8.0
頂點6，屬性Infinity，收到消息3.0，合并后的屬性3.0
//------------------------------------------ 第二次迭代 ------------------------------------------
頂點3 給 頂點2 發送消息 12.0
頂點2，屬性Infinity，收到消息12.0，合并后的屬性12.0
//------------------------------------------ 第三次迭代 ------------------------------------------
頂點2 給 頂點4 發送消息 14.0
頂點2 給 頂點1 發送消息 19.0
頂點1，屬性Infinity，收到消息19.0，合并后的屬性19.0
頂點4，屬性Infinity，收到消息14.0，合并后的屬性14.0
//------------------------------------------ 第四次迭代 ------------------------------------------
頂點4 給 頂點1 發送消息 15.0
頂點1，屬性19.0，收到消息15.0，合并后的屬性15.0
//------------------------------------------ 第五次迭代不用發送消息 ------------------------------------------

總結

以上是生活随笔為你收集整理的大数据基础之Spark——Spark pregel详细过程，一看就懂的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。