SparkGraphX中的PR算法和pregel迭代算法

PR算法（佩奇等级）

PR算法是早期构建搜索系统的链接分析算法，用于衡量特定网页相对于搜索引擎索引中其他网页而言的重要程度

一个页面的PR值越高，则对于其他网页则越重要

如图：

 由图通过迭代公式Vn=T·Vn-1,得到一个稳定的PR，矩阵如下：

但是有的点只有入度，没有出度，或者存在自环现象，引入公式Vn=αTVn-1+(1-α)V0则有几率摆脱极端现象

术语：

PageRank是执行多次连接的一个迭代算法，因此它是RDD分区操作的一个很好的用例。算法会维护两个数据集：一个由（pageID，linkList）的元素组成，包含每个页面的相邻页面的列表；
另一个由（pageID，rank）元素组成，包含每个页面的当前排序值。它按如下步骤进行计算。
1.将每个页面的排序值初始化为1.0。
2.在每次迭代中，对页面p，向其每个相邻页面（有直接链接的页面）发送一个值为rank(p)/numNeighbors(p)的贡献值。
3.将每个页面的排序值设为0.15 + 0.85 * contributionsReceived。

PR(tol,resetProb)，阻尼值一般默认为0.85(经验值)，tol可以设置，为迭代级别

object MyPageRank {
  def main(args: Array[String]): Unit = {
    val spark=SparkSession.builder().master("local[*]").appName("job").getOrCreate()
    val sc = spark.sparkContext
    //创建所有点
    val points = Seq((1L, ("Alice", 28)), (2L, ("Bob", 27)), (3L, ("Charlie", 65)),
      (4L, ("Doinb", 42)), (5L, ("Ed", 55)), (6L, ("Faker", 50)))
    //创建所有的边
    val eds = Seq(Edge(2L,1L,7),Edge(2L,4L,2),Edge(4L,1L,1),Edge(5L,2L,2),Edge(5L,3L,8),
      Edge(5L,6L,3),Edge(3L,2L,4),Edge(3L,6L,3))
    val vertices=sc.makeRDD(points)
    val edges=sc.makeRDD(eds)
    val graph=Graph(vertices,edges) //spark graph创建成功
    //PR算法  用于衡量特定网页相对于搜索引擎索引中的其他网页而言的重要程度
    val ranks = graph.pageRank(0.00001)
    ranks.vertices.sortBy(_._2,false).collect().foreach(println)

    spark.stop()
  }
}

(1,1.7924127957615186)
(2,0.9969646507526428)
(6,0.9969646507526428)
(4,0.9688717814927128)
(3,0.6996243163176442)
(5,0.5451618049228396)

Pregel算法

Pregel作为分布式图计算的计算框架，主要用于图遍历、最短路径、PageRank计算等等

object MyPregel {
  def main(args: Array[String]): Unit = {
    //找图中最小值  pregel迭代算法
    val spark=SparkSession.builder().master("local[*]").appName("job").getOrCreate()
    val sc =spark.sparkContext
    val vertices = sc.parallelize(Array((1L,(7,-1)),(2L,(3,-1)),(3L,(2,-1)),(4L,(6,-1))))
    val edges = sc.parallelize(Array(Edge(1L,2L,true),Edge(1L,4L,true),Edge(2L,4L,true),Edge(3L,1L,true),Edge(3L,4L,true)))
    val graph = Graph(vertices,edges)

    var initMsg = 999
    def vprog(vid:VertexId,value:(Int,Int),msg:Int)={
      if (msg==initMsg) value else(msg min value._1,value._1)
    }
    def sendMsg(triplet:EdgeTriplet[(Int,Int),Boolean])={
      val sa = triplet.srcAttr
      if (sa._1==sa._2) Iterator.empty else Iterator((triplet.dstId,sa._1))
    }
    def mergeMsg(msg1:Int,msg2:Int)={
      msg1 min msg2
    }

    val tg = graph.pregel(initMsg, Int.MaxValue, EdgeDirection.Out)(vprog, sendMsg, mergeMsg)
    tg.vertices.collect().foreach(println)
    spark.stop()

  }
}