189 8069 5689
今天就跟大家聊聊有关spark-shell如何实现PageRank，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。
成都创新互联主营巴彦网站建设的网络公司,主营网站建设方案,成都App定制开发,巴彦h5微信小程序开发搭建,巴彦网站营销推广欢迎巴彦等地区企业咨询
talk is cheap, show code
Scala 代码实现 :
val links = sc.parallelize(Array(
('a', Array('d')),
('b', Array('a')),
('c', Array('a', 'b')),
('d', Array('a', 'c'))))
// 设置各页面的初始rank值为1.0
// var ranks = links.mapValues(_ => 1.0)
var ranks = sc.parallelize(Array(
('a', 1.0),
('b', 1.0),
('c', 1.0),
('d', 1.0)))
for(i <- 1 to iterations) {
val joinRdd = links.join(ranks)
val contribsRdd = joinRdd.flatMap{
// 注意这里的links为模式匹配得到的值, 类型为Array[Char], 并非前面的ParallelCollectionRDD
case(srcURL, (links, rank)) => links.map(destURL => (destURL, rank / links.size))
}
// 简化了的 rank 计算公式, 更新 ranks
ranks = contribsRdd.reduceByKey(_ + _).mapValues(0.15 + _ * 0.85)
}代码分析:
第1次迭代, transform操作, rdd.join(other) , 对 links 和 ranks 这个2个RDD进行连接
joinRdd = links.join(ranks)
结果如下:
res1: Array[(Char, (Array[Char], Double))] = Array((d,(Array(a, c),1.0)), (b,(Array(a),1.0)), (a,(Array(d),1.0)), (c,(Array(a, b),1.0)))
第1次迭代, transform操作: rdd.flatMap(func)
contribsRdd = joinRdd.flatMap{
// 注意这里的links为模式匹配得到的值, 类型为Array[Char],并非前面的ParallelCollectionRDD
case(srcURL, (links, rank)) => links.map(destURL => (destURL, rank / links.size))
}
结果如下:
res2: Array[(Char, Double)] = Array((a,0.5), (c,0.5), (a,1.0), (d,1.0), (a,0.5), (b,0.5))
第1次迭代, transform操作: rdd.reduceByKey(func) & rdd.mapValues(func)
// 简化了的 rank 计算公式, 更新 ranks
ranks = contribsRdd.reduceByKey(_ + _).mapValues(0.15 + _ * 0.85)
结果如下:
res3: Array[(Char, Double)] = Array((d,1.0), (b,0.575), (a,1.84999), (c,0.575))
这里有double类型造成的精度丢失, a 页面的 rank 值应该为 1.85
迭代的结果:
第1次迭代, ranks的结果如下:
res1: Array[(Char, Double)] = Array((d,1.0), (b,0.575), (a,1.84999), (c,0.575))
第2次迭代, ranks的结果如下:
res2: Array[(Char, Double)] = Array((d,1.72249), (b,0.394375), (a,1.308124), (c,0.575))
第3次迭代, ranks的结果如下:
res3: Array[(Char, Double)] = Array((d,1.26190), (b,0.39437), (a,1.46165), (c,0.88206))
...
第21次迭代, ranks的结果如下:
res21: Array[(Char, Double)] = Array((d,1.37039), (b,0.46126), (a,1.43586), (c,0.73247))
第22次迭代, ranks的结果如下:
res22: Array[(Char, Double)] = Array((d,1.37048), (b,0.46130), (a,1.43579), (c,0.73241))
第23次迭代, ranks的结果如下:
res23: Array[(Char, Double)] = Array((d,1.37042), (b,0.46127), (a,1.43583), (c,0.73245))
从上面的迭代结果可以得出, 当迭代到22次时, rank值( 保留小数点后4位 )趋于稳定,
此时的rank值可以作为结果值:
a : 1.4358 b : 0.4613 c : 0.7325 d : 1.3704
看完上述内容，你们对spark-shell如何实现PageRank有进一步的了解吗？如果还想了解更多知识或者相关内容，请关注创新互联行业资讯频道，感谢大家的支持。