一.GraphX 介绍
图
图可以用来表达自然连接关系,,比如:社交网络、移动网络、股份关系等等。
主要由点和边组成,是以“图论”为基础抽象表示关联关系。
GraphX
GraphX是一个Spark组件,用于图的计算,引入新的模型来扩展RDD,属性信息包含点和边,扩展一些API,简化图计算。
特点
灵活性高,图和集合无缝连接,RDD分布式计算可以做的事情,GraphX也可以完成,有效互相转换,运用Pregel 编写自定义迭代计算。
速度快,可以跟Spark很好的结合。
算法多,除了高度灵活的API之外,GraphX还提供了各种图形算法,其中许多都是由我们的用户提供的。PageRank,Pregel,连接组件,标签传播,SVD ++,强大的连接组件以及三角计数等。
二.GraphX 实现分析
2.1 图的切分方式
图切分总体来说点切分和边切分。
(图片来源官网)
从社区划分角度,例如:明星结点可以跟很多结点去产生关系,边的关系会产生很多。
边切分:如果用此方式去切分,结点信息保存完整,边的信息不完整,就需要迁移大量结点去管理,需要消耗大量资源去通信,每个机器服务器的副本数据增加。
点切分:如果用此方式去切分,关联关系信息完整,只需要把缺失的结点信息通信复制过去,需要耗费资源少,尤其针对边的关系特别多的社交网络,这种方式优化资源情况好,GraphX 用的是这种方式。
2.2 数据处理
Spark经历过很多版本,核心抽象一直是RDD,表示一个可变的,可分区,可并行的元素数据集合,其他的属性操作可以跟RDD对接使用,所有的计算操作都是基于RDD。
2.3 BSP模型
图计算模型基本遵循分布式批量同步BSP计算模型,目前有两种比较成熟的图计算框架:Pregel框架和GraphLab框架。
BSP(Bulk Synchronous Parallel Computing Model,块同步并行计算模型,又称“大同步”模型)计算过程包括一系列全局超步(所谓的超步就是计算中的一次迭代),每个超步主要包括三个组件:
局部计算:每个参与的处理器都有自身的计算任务。
全局tpn:处理器群相互交换数据。
栅栏同步(Barrier Synchronization):当一个处理器遇到“路障”(或栅栏),会等到其他所有处理器完成它们的计算步骤。
2.4 设计核心
图视图操作,转换成其关联Table(顶点,边,路由)的RDD操作完成,一个图计算在逻辑上等价成一系列RDD操作,因此Graph继承了RDD的特性:不变性、并行、容错,逻辑上图操作转换成一个新图,会有一系列优化操作。
索引结构在图的转换过程中是共用的,降低存储和计算资源。
图的分布式存储方式是点切分,可以使用分区算子进行指定划分。
三.GraphX 实例
3.1 创建
def apply[VD: ClassTag, ED: ClassTag]
根据边直接创建, 所有顶点的属性都一样为 defaultValue
def fromEdges[VD: ClassTag, ED: ClassTag]
def fromEdgeTuples[VD: ClassTag]
3.2 转换操作
3.2.1 基本信息
获取边的数量
val numEdges: Long
获取顶点的数量
val numVertices: Long
获取所有顶点的入度
val inDegrees: VertexRDD[Int]
获取所有顶点的出度
val outDegrees: VertexRDD[Int]
获取所有顶点入度与出度之和
val degrees: VertexRDD[Int]
查看所有边
val edges: EdgeRDD[ED]
所有顶点属性
val vertices: VertexRDD[VD]
边和顶点属性(包含id)
val triplets: RDD[EdgeTriplet[VD, ED]]
3.2.2 mapVertices
用于更新节点属性信息
def mapVerticesVD2: ClassTag => VD2)
(implicit eq: VD =:= VD2 = null): Graph[VD2, ED]
//创建顶点数据集
val vertexRDD:RDD[(VertexId,(String,String))] = context.makeRDD(Array(
(3L,("zhangsan","student")),
(7L,("limeng","学生")),
(5L,("lisi","老师")),
(2L,("wangwu","老师")),
(4L,("zhaoliu","老师")),
(6L,("wangguo","老师"))
))
//创建边的数据
val edgesRDD:RDD[Edge[String]] = context.makeRDD(Array(
Edge(3L,7L,"测试关系"),
Edge(5L,3L,"测试指导"),
Edge(2L,5L,"测试朋友"),
Edge(5L,7L,"测试朋友")
))
//构建一个图
val graphx = Graph(vertexRDD,edgesRDD)
println("mapVertices: ")
graphx.mapVertices((id,a) => (a._1,a.2))
graphx.vertices.collect().foreach(println())
graphx.edges.collect().foreach(println(_))
结果
mapVertices:
(4,(zhaoliu,老师))
(6,(wangguo,老师))
(3,(zhangsan,student))
(7,(limeng,学生))
(5,(lisi,老师))
(2,(wangwu,老师))
Edge(2,5,测试朋友)
Edge(3,7,测试关系)
Edge(5,3,测试指导)
Edge(5,7,测试朋友)
3.2.3 mapEdges
用于图的边属性修改,方向不变。
graphx.mapEdges(attr=>{
1
}) .triplets
.collect
.foreach(triplet => println(s"srcId=${triplet.srcId} srcAttr=${triplet.srcAttr}--edge=${triplet.attr}--dstId=${triplet.dstId} dstAttr=${triplet.dstAttr} "))
结果
srcId=2 srcAttr=(wangwu,老师)--edge=1--dstId=5 dstAttr=(lisi,老师)
srcId=3 srcAttr=(zhangsan,student)--edge=1--dstId=7 dstAttr=(limeng,学生)
srcId=5 srcAttr=(lisi,老师)--edge=1--dstId=3 dstAttr=(zhangsan,student)
srcId=5 srcAttr=(lisi,老师)--edge=1--dstId=7 dstAttr=(limeng,学生)
3.2.4 mapTriplets
修改边的属性。
def mapTripletsED2: ClassTag: Graph[VD, ED2] = {
mapTriplets((pid, iter) => iter.map(map), TripletFields.All)
}
graphx.mapTriplets(e=>{
s"${e.srcAttr}:${e.dstAttr}:${e.attr}"
}).triplets
.collect
.foreach(triplet => println(s"srcId=${triplet.srcId} srcAttr=${triplet.srcAttr} edge=${triplet.attr} dstId=${triplet.dstId} dstAttr=${triplet.dstAttr} "))
结果
srcId=2 srcAttr=(wangwu,老师) edge=(wangwu,老师):(lisi,老师):测试朋友 dstId=5 dstAttr=(lisi,老师)
srcId=3 srcAttr=(zhangsan,student) edge=(zhangsan,student):(limeng,学生):测试关系 dstId=7 dstAttr=(limeng,学生)
srcId=5 srcAttr=(lisi,老师) edge=(lisi,老师):(zhangsan,student):测试指导 dstId=3 dstAttr=(zhangsan,student)
srcId=5 srcAttr=(lisi,老师) edge=(lisi,老师):(limeng,学生):测试朋友 dstId=7 dstAttr=(limeng,学生)
3.3 结构操作
3.3.1 reverse
图的反转操作转换成一个新图,所有边的方向都是反向。
println("mapVertices :")
graphx.mapVertices((id,attr)=>attr._1)
.triplets
.collect
.foreach(triplet => println(s"srcId=${triplet.srcId} srcAttr=${triplet.srcAttr} edge=${triplet.attr} dstId=${triplet.dstId} dstAttr=${triplet.dstAttr} "))
println("reverse: ")
graphx.reverse.triplets
.collect
.foreach(triplet => println(s"srcId=${triplet.srcId} srcAttr=${triplet.srcAttr} edge=${triplet.attr} dstId=${triplet.dstId} dstAttr=${triplet.dstAttr} "))
结果
mapVertices :
srcId=2 srcAttr=wangwu edge=测试朋友 dstId=5 dstAttr=lisi
srcId=3 srcAttr=zhangsan edge=测试关系 dstId=7 dstAttr=limeng
srcId=5 srcAttr=lisi edge=测试指导 dstId=3 dstAttr=zhangsan
srcId=5 srcAttr=lisi edge=测试朋友 dstId=7 dstAttr=limeng
reverse:
srcId=3 srcAttr=(zhangsan,student) edge=测试指导 dstId=5 dstAttr=(lisi,老师)
srcId=5 srcAttr=(lisi,老师) edge=测试朋友 dstId=2 dstAttr=(wangwu,老师)
srcId=7 srcAttr=(limeng,学生) edge=测试关系 dstId=3 dstAttr=(zhangsan,student)
srcId=7 srcAttr=(limeng,学生) edge=测试朋友 dstId=5 dstAttr=(lisi,老师)
3.3.2 subgraph(重点)
划分子图,epred 边属性判断,vpred 顶点属性判断
def subgraph(
epred: EdgeTriplet[VD, ED] => Boolean = (x => true),
vpred: (VertexId, VD) => Boolean = ((v, d) => true))
: Graph[VD, ED]
graphx.subgraph(x=>if(x.attr=="测试朋友") true else false,(id,attr)=>true).triplets
.collect
.foreach(triplet => println(s"srcId=${triplet.srcId} srcAttr=${triplet.srcAttr} edge=${triplet.attr} dstId=${triplet.dstId} dstAttr=${triplet.dstAttr} "))
结果
srcId=2 srcAttr=(wangwu,老师) edge=测试朋友 dstId=5 dstAttr=(lisi,老师)
srcId=5 srcAttr=(lisi,老师) edge=测试朋友 dstId=7 dstAttr=(limeng,学生)
3.3.3 mask
创建一个子图,子图的范围other中,取交集
def maskVD2: ClassTag, ED2: ClassTag: Graph[VD, ED]
val validGraph = graphx.subgraph(vpred = (id, attr) => attr._2 != "student")
graphx.mask(validGraph).triplets
.collect
.foreach(triplet => println(s"srcId=${triplet.srcId} srcAttr=${triplet.srcAttr} edge=${triplet.attr} dstId=${triplet.dstId} dstAttr=${triplet.dstAttr} "))
结果
srcId=2 srcAttr=(wangwu,老师) edge=测试朋友 dstId=5 dstAttr=(lisi,老师)
srcId=5 srcAttr=(lisi,老师) edge=测试朋友 dstId=7 dstAttr=(limeng,学生)
3.3.4 groupEdges
根据边分组,边如果在同一个分区,才可以划分成新图,必须使用partitionBy重分区
def groupEdges(merge: (ED, ED) => ED): Graph[VD, ED]
graphx.partitionBy(PartitionStrategy.RandomVertexCut,1).groupEdges((e1,e2)=>e1+e2).triplets.collect().foreach(triplet=>println(s"srcId=${triplet.srcId} srcAttr=${triplet.srcAttr} edge=${triplet.attr} dstId=${triplet.dstId} dstAttr=${triplet.dstAttr} "))
结果
srcId=2 srcAttr=(wangwu,老师) edge=测试朋友 dstId=5 dstAttr=(lisi,老师)
srcId=3 srcAttr=(zhangsan,student) edge=测试关系 dstId=7 dstAttr=(limeng,学生)
srcId=5 srcAttr=(lisi,老师) edge=测试指导 dstId=3 dstAttr=(zhangsan,student)
srcId=5 srcAttr=(lisi,老师) edge=测试朋友,好的 dstId=7 dstAttr=(limeng,学生)
5->7 这个边属性有变化。
3.4 聚合操作
3.4.1 collectNeighbors
该方法的作用是收集每个顶点的邻居顶点的顶点id和顶点属性。需要指定方向
EdgeDirection.out:出的方向
EdgeDirection.in:入的方向
EdgeDirection.Either:入边或出边
EdgeDirection.Both:入边且出边
def collectNeighbors(edgeDirection: EdgeDirection): VertexRDD[Array[(VertexId, VD)]] 收集邻居节点的数据,根据指定的方向。返回的数据为RDD[(VertexId, Array[(VertexId, VD)])]
graphx.collectNeighbors(EdgeDirection.Either).collect().foreach(f=>{
val colls : Array[(VertexId, (String, String))] = f.2
println(s"VertexId:${f._1} ${colls.foreach(println())}")
})
结果
VertexId:4 ()
VertexId:6 ()
(7,(limeng,学生))
(5,(lisi,老师))
VertexId:3 ()
(3,(zhangsan,student))
(5,(lisi,老师))
VertexId:7 ()
(2,(wangwu,老师))
(3,(zhangsan,student))
(7,(limeng,学生))
VertexId:5 ()
(5,(lisi,老师))
VertexId:2 ()
3.4.2 collectNeighborIds
该方法的作用是收集每个顶点的邻居顶点的顶点id。它的实现和collectNeighbors非常相同。需要指定方向。
def collectNeighborIds(edgeDirection: EdgeDirection): VertexRDD[Array[VertexId]]
graphx.collectNeighborIds(EdgeDirection.Either).collect().foreach(f=>{
val colls :Array[(VertexId)] = f.2
println(s"VertexId:${f._1} ${colls.foreach(println())}")
})
结果
VertexId:4 ()
VertexId:6 ()
7
5
VertexId:3 ()
3
5
VertexId:7 ()
2
3
7
VertexId:5 ()
5
VertexId:2 ()
3.4.3 aggregateMessages
def aggregateMessagesA: ClassTag => A, tripletFields: TripletFields = TripletFields.All) : VertexRDD[A]
每一个边都会通过sendMsg 发送一个消息,每一个顶点都会通过mergeMsg 来处理所有他收到的消息。
TripletFields存在主要用于定制 EdgeContext 对象中的属性的值是否存在,为了减少数据通信量。
val myVertices: RDD[(Long, String)] = context.makeRDD(Array((1L,"Ann"),(2L,"Bill"),(3L,"Charles"),(4L,"Diane"),(5L,"Went")))
val myEdges: RDD[Edge[String]] = context.makeRDD(Array(Edge(1L,2L,"is"),Edge(2L,3L,"is"),Edge(3L,4L,"is"),Edge(4L,5L,"like"),Edge(3L,5L,"wrote")))
val myGraph: Graph[String, String] = Graph(myVertices,myEdges)
//出度
val vertices1:VertexRDD[Int] = myGraph.aggregateMessagesInt,+)
println("vertices1:")
vertices1.collect().foreach(println(_))
结果
vertices1:
(4,1)
(3,1)
(5,2)
(2,1)
3.5 关联操作
3.5.1 joinVertices (重点)
def joinVerticesU: ClassTag])(mapFunc: (VertexId, VD, U) => VD) : Graph[VD, ED]
一一对应取交集,类型不变
val users: RDD[(VertexId, (String, String))] =
context.parallelize(Array((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),
(5L, ("franklin", "prof")), (2L, ("istoica", "prof"))))
val relationships: RDD[Edge[(String,String)]] =
context.parallelize(Array(Edge(3L, 7L, ("collab","testcollab")), Edge(5L, 3L, ("advisor","testadvisor")),
Edge(2L, 5L, ("colleague","testcolleague")), Edge(5L, 7L, ("pi","testpi"))))
val graph = Graph(users, relationships)
//使用joinVertices操作,用user2中的属性替换图中对应Id的属性 先将图中的顶点属性置空
val users2: RDD[(VertexId, (String, Int))] =
context.parallelize(Array((3L, ("zzzrxin", 30)), (7L, ("zzzzjgonzal", 70)),
(5L, ("zzzfranklin", 50)), (2L, ("zzzistoica", 20))))
//取交集,结点没有置空
println("joinVertices")
val joinVerticesValue = graph.mapVertices((id,attr) => "").joinVertices(users2){(vid,empty,user) =>user.2.toString}
joinVerticesValue.vertices.collect().foreach(println())
结果
(2,20)
(3,30)
(5,50)
(7,70)
3.5.2 outerJoinVertices (重点)
def outerJoinVerticesU: ClassTag, VD2: ClassTag])
(mapFunc: (VertexId, VD, Option[U]) => VD2)(implicit eq: VD =:= VD2 = null)
: Graph[VD2, ED]
跟joinVertices类似,如果没有相对应的结点,赋值。
val vertexRDD3 = sc.makeRDD(Array(
(1L,("zhang1",25)),
(2L,("zhang2",18)),
(3L,("zhang3",45)),
(4L,("zhang4",20)),
(5L,("zhang5",70)),
(6L,("zhang6",79)),
(7L,("zhang7",101))
))
val edgesRDD3=sc.makeRDD(Array(
Edge(1L,2L,"1"),
Edge(3L,2L,"1"),
Edge(1L,4L,"1"),
Edge(7L,2L,"1"),
Edge(3L,5L,"1"),
Edge(7L,6L,"1"),
Edge(6L,4L,"1")
))
val graph3: Graph[(String, Int), String] = Graph(vertexRDD3, edgesRDD3)
//取交集,结点没有赋值null
println("outerJoinVertices:")
graph3.outerJoinVertices(vertexRDD4)((v, vd, u) => {
u.getOrElse(null)
}).triplets.collect().foreach(triplet=>{
println(s"srcId=${triplet.srcId} srcAttr=${triplet.srcAttr} --edge=${triplet.attr} --dstId=${triplet.dstId} dstAttr=${triplet.dstAttr} ")
})
结果
outerJoinVertices:
srcId=1 srcAttr=(test1,61) --edge=1 --dstId=2 dstAttr=(test2,62)
srcId=3 srcAttr=(test3,63) --edge=1 --dstId=2 dstAttr=(test2,62)
srcId=1 srcAttr=(test1,61) --edge=1 --dstId=4 dstAttr=null
srcId=7 srcAttr=null --edge=1 --dstId=2 dstAttr=(test2,62)
srcId=3 srcAttr=(test3,63) --edge=1 --dstId=5 dstAttr=null
srcId=7 srcAttr=null --edge=1 --dstId=6 dstAttr=null
srcId=6 srcAttr=null --edge=1 --dstId=4 dstAttr=null
3.6 Pregel
def pregelA: ClassTag(
//结点调用该消息将聚合后的数据和本结点进行属性的合并
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)
}
pregel模型基于BSP实现并行计算,分步骤计算,所有SuperStep(超步)顶点都是并行,安按照发送消息方向聚合到边上,归集属性,根据迭代次数判断输出结果。
图片: https://uploader.shimo.im/f/TcPGr0A6puDd3kBW.png!thumbnail?accessToken=eyJhbGciOiJIUzI1NiIsImtpZCI6ImRlZmF1bHQiLCJ0eXAiOiJKV1QifQ.eyJleHAiOjE2ODk0NzcyODUsImZpbGVHVUlEIjoicldUUHJRWDhYanRSV1FZMyIsImlhdCI6MTY4OTQ3Njk4NSwiaXNzIjoidXBsb2FkZXJfYWNjZXNzX3Jlc291cmNlIiwidXNlcklkIjoyMzg0NzMxMn0.M5AW0LPOn6zzTqo4pBcm2YLfngACLgpeDv6gtvqybUg
3.6.1 最短路径
//最短路径算法:计算的是从V1到所有顶点的最短路径
val vertexRDD5:RDD[(VertexId,Int)] = sc.makeRDD(Array(
(1L,0),
(2L,Int.MaxValue),
(3L,Int.MaxValue),
(4L,Int.MaxValue),
(5L,Int.MaxValue),
(6L,Int.MaxValue),
(7L,Int.MaxValue),
(8L,Int.MaxValue),
(9L,Int.MaxValue)
))
val edgesRDD5 = sc.makeRDD(Array(
Edge(1L,2L,6),
Edge(1L,3L,3),
Edge(1L,4L,1),
Edge(3L,2L,2),
Edge(3L,4L,2),
Edge(2L,5L,1),
Edge(5L,4L,6),
Edge(5L,6L,4),
Edge(6L,5L,10),
Edge(5L,7L,3),
Edge(5L,8L,6),
Edge(4L,6L,10),
Edge(6L,7L,2),
Edge(7L,8L,4),
Edge(9L,5L,2),
Edge(9L,8L,3)
))
val graph5: Graph[Int, Int] = Graph(vertexRDD5, edgesRDD5)
val pathGraph: Graph[Int, Int] = graph5.pregelInt( vprog = (vid,a,b) =>math.min(a,b),triplet=>{
//如果三元组的边的权重+入口的值小于目的的值,那么就发送消息,反之不发送
if(triplet.srcAttr!= Int.MaxValue && triplet.srcAttr+triplet.attrmath.min(a,b))
println("vertices")
pathGraph.vertices.collect().foreach(println(_))
结果
vertices
(8,12)
(1,0)
(9,2147483647)
(2,5)
(3,3)
(4,1)
(5,6)
(6,10)
(7,9)
分析
这个算法有个局限,无法把全链路展示出来,比如到达结点8,中间的结点无法展示,结果集第二列是所有边加起来的权值。
3.6.2 进化版最短路径
这个版本算法,为了把全链路展示出来,显示到达结点的最短路径。
val v = sc.makeRDD(Array((1L,"A"),(2L,"B"),(3L,"C"),(4L,"D"),(5L,"E"),(6L,"F"),(7L,"G")))
val e =sc.makeRDD(Array(Edge(1L,2L,7.0),Edge(1L,4L,5.0),Edge(2L,3L,8.0),Edge(2L,4L,9.0),
Edge(2L,5L,7.0),Edge(3L,5L,5.0),Edge(4L,5L,15.0),Edge(4L,6L,6.0),Edge(5L,6L,8.0)
,Edge(5L,7L,9.0),Edge(6L,7L,11.0)))
val graph: Graph[String, Double] = Graph(v, e)
println("dijkstra pregel 最短路径+距离")
dijkstra2(graph,1L).vertices.collect().foreach(println(_))
def dijkstra2(g:Graph[String,Double],origin:VertexId) ={
var g2: Graph[(Boolean, Double, List[VertexId]), Double] = g.mapVertices((vid, _) => {
(false, if (vid == origin) 0 else Double.MaxValue, ListVertexId)
})
//遍历所有点
(0L until g.vertices.count()).foreach(f=>{
val currentVerID = g2.vertices.filter(!_._2._1).fold(0L,(false,Double.MaxValue,ListVertexId))((a,b)=>{
if(a._2._2 < b._2._2) a else b
})._1
//当前结点的最短距离列表
val newDistance: VertexRDD[(Double,List[VertexId])] =g2.aggregateMessages(Double,List[VertexId]) {
ctx.sendToDst((ctx.srcAttr._2 + ctx.attr, ctx.srcAttr._3 :+ ctx.srcId))
}
},
(a,b)=> if(a._1 < b._1) a else b
)
//比较记录权值与获取列表,找出最小权重
g2 = g2.outerJoinVertices(newDistance){
(vid,vd,newSum) => {
val newSumVal = newSum.getOrElse((Double.MaxValue,ListVertexId))
((vd._1 || vid == currentVerID),
math.min(vd._2,newSumVal._1),
if(vd._2 < newSumVal._1 ) vd._3 else newSumVal._2
)}
}
})
//所有结点遍历完毕,获取最短路径+链路
val result: Graph[(String, Double, List[VertexId]), Double] = g.outerJoinVertices(g2.vertices) {
(vid, vd, dist) => {
(vd, dist.getOrElse((false, Double.MaxValue, ListVertexId))._2, dist.getOrElse((false, Double.MaxValue, ListVertexId))._3)
}
}
result
}
结果
dijkstra pregel 最短路径+距离
(1,(A,0.0,List()))
(2,(B,7.0,List(1)))
(3,(C,15.0,List(1, 2)))
(4,(D,5.0,List(1)))
(5,(E,14.0,List(1, 2)))
(6,(F,11.0,List(1, 4)))
(7,(G,22.0,List(1, 4, 6)))
3.7 PageRank算法
3.7.1 原理
PageRank让链接来"投票"
一个页面的“得票数”由所有链向它的页面的重要性来决定,到一个页面的超链接相当于对该页投一票。一个页面的PageRank是由所有链向它的页面(“链入页面”)的重要性经过递归算法得到的。一个有较多链入的页面会有较高的等级,相反如果一个页面没有任何链入页面,那么它没有等级。
2005年初,Google为网页链接推出一项新属性nofollow,使得网站管理员和网站作者可以做出一些Google不计票的链接,也就是说这些链接不算作"投票"。nofollow的设置可以抵制评论垃圾。
假设一个由4个页面组成的小团体:A,B,C和D。如果所有页面都链向A,那么A的PR(PageRank)值将是B,C及D的Pagerank总和。
图片: https://uploader.shimo.im/f/0lXNrCyaqvQLI9Ea.png!thumbnail?accessToken=eyJhbGciOiJIUzI1NiIsImtpZCI6ImRlZmF1bHQiLCJ0eXAiOiJKV1QifQ.eyJleHAiOjE2ODk0NzcyODUsImZpbGVHVUlEIjoicldUUHJRWDhYanRSV1FZMyIsImlhdCI6MTY4OTQ3Njk4NSwiaXNzIjoidXBsb2FkZXJfYWNjZXNzX3Jlc291cmNlIiwidXNlcklkIjoyMzg0NzMxMn0.M5AW0LPOn6zzTqo4pBcm2YLfngACLgpeDv6gtvqybUg
继续假设B也有链接到C,并且D也有链接到包括A的3个页面。一个页面不能投票2次。所以B给每个页面半票。以同样的逻辑,D投出的票只有三分之一算到了A的PageRank上。
图片: https://uploader.shimo.im/f/LcrfEhM3amjijiL2.png!thumbnail?accessToken=eyJhbGciOiJIUzI1NiIsImtpZCI6ImRlZmF1bHQiLCJ0eXAiOiJKV1QifQ.eyJleHAiOjE2ODk0NzcyODUsImZpbGVHVUlEIjoicldUUHJRWDhYanRSV1FZMyIsImlhdCI6MTY4OTQ3Njk4NSwiaXNzIjoidXBsb2FkZXJfYWNjZXNzX3Jlc291cmNlIiwidXNlcklkIjoyMzg0NzMxMn0.M5AW0LPOn6zzTqo4pBcm2YLfngACLgpeDv6gtvqybUg
换句话说,根据链出总数平分一个页面的PR值。
图片: https://uploader.shimo.im/f/aFPIJC7EC7HVlBdR.png!thumbnail?accessToken=eyJhbGciOiJIUzI1NiIsImtpZCI6ImRlZmF1bHQiLCJ0eXAiOiJKV1QifQ.eyJleHAiOjE2ODk0NzcyODUsImZpbGVHVUlEIjoicldUUHJRWDhYanRSV1FZMyIsImlhdCI6MTY4OTQ3Njk4NSwiaXNzIjoidXBsb2FkZXJfYWNjZXNzX3Jlc291cmNlIiwidXNlcklkIjoyMzg0NzMxMn0.M5AW0LPOn6zzTqo4pBcm2YLfngACLgpeDv6gtvqybUg
最后,所有这些被换算为一个百分比再乘上一个系数。由于“没有向外链接的页面”传递出去的PageRank会是0,所以,Google通过数学系统给了每个页面一个最小值:
图片: https://uploader.shimo.im/f/leBsW4KIXw7spnHa.png!thumbnail?accessToken=eyJhbGciOiJIUzI1NiIsImtpZCI6ImRlZmF1bHQiLCJ0eXAiOiJKV1QifQ.eyJleHAiOjE2ODk0NzcyODUsImZpbGVHVUlEIjoicldUUHJRWDhYanRSV1FZMyIsImlhdCI6MTY4OTQ3Njk4NSwiaXNzIjoidXBsb2FkZXJfYWNjZXNzX3Jlc291cmNlIiwidXNlcklkIjoyMzg0NzMxMn0.M5AW0LPOn6zzTqo4pBcm2YLfngACLgpeDv6gtvqybUg
说明:在Sergey Brin和Lawrence Page的1998年原文中给每一个页面设定的最小值是1-d,而不是这里的(1-d)/N。 所以一个页面的PageRank是由其他页面的PageRank计算得到。Google不断的重复计算每个页面的PageRank。如果给每个页面一个随机PageRank值(非0),那么经过不断的重复计算,这些页面的PR值会趋向于稳定,也就是收敛的状态。这就是搜索引擎使用它的原因。
引用:https://baike.baidu.com/item/google%20pagerank/2465380?fromtitle=pagerank&fromid=111004
3.7.2 实现
退出方式
参数值越小得到的结果越有说服力。
固定的迭代次数(静态)与公差的退出方式(动态),对于这两种调用方式(面向对象和基于对象)。
静态退出方式传入numIter参数(迭代次数),动态退出方式传入tol参数(公差)。如果全图的顶点变化值全部小于tol,算法终止。
GraphX用到resetProb与PageRank论文中1-d意思相同,d表示抑制因子。PageRank论文中推荐抑制因子0.85,GraphX文档中推荐resetProb为0.15。
val edge1: RDD[Edge[Int]] = sc.makeRDD(Array(
Edge(2, 1, 0),
Edge(4, 1, 0),
Edge(1, 2, 0),
Edge(6, 3, 0),
Edge(7, 3, 0),
Edge(7, 6, 0),
Edge(6, 7, 0),
Edge(3, 7, 0)
))
val vertexValue: RDD[(VertexId, String)] = sc.makeRDD(Array(
(1L, "ba1"),
(2L, "ba2"),
(3L, "ba3"),
(4L, "ba4"),
(6L, "ba6"),
(7L, "ba7"),
(8L, "ba8")
))
val graph1: Graph[Int, Int] = Graph.fromEdges(edge1,0)
val graph2: Graph[Double, Double] = graph1.pageRank(0.0001)
graph2.triplets
.collect
.foreach(triplet => println(s"srcId=${triplet.srcId} srcAttr=${triplet.srcAttr}--edge=${triplet.attr}--dstId=${triplet.dstId} dstAttr=${triplet.dstAttr} "))
结果
srcId=2 srcAttr=1.3907556008752426--edge=1.0--dstId=1 dstAttr=1.4596227918476916
srcId=4 srcAttr=0.15007622780470478--edge=1.0--dstId=1 dstAttr=1.4596227918476916
srcId=1 srcAttr=1.4596227918476916--edge=1.0--dstId=2 dstAttr=1.3907556008752426
srcId=6 srcAttr=0.7017164142469724--edge=0.5--dstId=3 dstAttr=0.9998520559494657
srcId=7 srcAttr=1.2979769092759237--edge=0.5--dstId=3 dstAttr=0.9998520559494657
srcId=7 srcAttr=1.2979769092759237--edge=0.5--dstId=6 dstAttr=0.7017164142469724
srcId=6 srcAttr=0.7017164142469724--edge=0.5--dstId=7 dstAttr=1.2979769092759237
srcId=3 srcAttr=0.9998520559494657--edge=1.0--dstId=7 dstAttr=1.2979769092759237