具体PageRank 算法原理 这里不多讲了，重点说下Spark 实现的过程
- 1 读入数据点 links sc.textfile ,split reduceByKey, 注意去重
- 2 初始化Rank
- 3 循环迭代：
- 3.1 利用 links join ranks 得到（List(点)，rank）
- 3.2 利用 flatMap 求得每个点真正的rank
- 3.3 利用reduce（,）聚合结果得到，最终结果
package org.apache.spark.myExamples
import org.apache.spark.sql.SparkSession
/**
* Created by wjf on 16-8-30.
*/
object TestPageRank {
def main(args: Array[String]): Unit = {
val spark =SparkSession.builder().appName("test PageRank").master("local").getOrCreate()
val iter=10
val links = spark.read.textFile("data/mllib/pagerank_data.txt").rdd.map { line =>
val parts = line.split(" ")
(parts(0), parts(1))
}.distinct().groupByKey().cache()
links.foreach(println)
var ranks = links.mapValues( v => 1.0)
ranks.foreach(println)
for(i <- 1 to iter){
val contri=links.join(ranks).values.flatMap{ case(urls,rank) =>
val size=urls.size
urls.map(url => (url,rank/size))
}
ranks=contri.reduceByKey(_ + _).mapValues(0.15 + 0.85*_)
println("Iteration %d".format(i))
ranks.foreach(println)
}
}
}
