How to implement PageRank with Spark
Spark how to achieve PageRank, I believe that many inexperienced people do not know what to do, so this paper summarizes the causes of the problem and solutions, through this article I hope you can solve this problem.
Brief introduction of PageRank algorithm
PageRank is an iterative algorithm that performs multiple joins, so it is a good use case for RDD partitioning operations. The algorithm maintains two datasets: one consists of (pageID,linkList) elements containing a list of adjacent pages for each page, and the other consists of (pageID,rank) elements containing the current ranking values for each page. It is calculated as follows.
Initialize the sort value for each page to 1.0.
In each iteration, for page p, a contribution value of rank (p) / numNeighbors (p) is sent to each of its adjacent pages (pages with direct links).
Set the sort value for each page to 0.15 + 0.85 * contributionsReceived.
The last two steps repeat several loops, and in the process, the algorithm gradually converges to the actual PageRank value of each page. In practice, convergence usually requires about 10 rounds of iterations.
Simulated data
Suppose a small group of four pages: a _ B _ _ C and D. The adjacent pages are as follows:
A:B C
B:A C
C:A B D
D:C
Object SparkPageRank {
Def showWarning () {
System.err.println (
WARN: This is a naive implementation of PageRank and is given as an example!
| | Please use the PageRank implementation found in org.apache.spark.graphx.lib.PageRank |
| for more conventional use.
"" .stripMargin)
}
Def main (args: Array [String]) {
If (args.length
< 1) { System.err.println("Usage: SparkPageRank ") System.exit(1) } showWarning() val spark = SparkSession .builder .appName("SparkPageRank") .getOrCreate() val iters = if (args.length >1) args (1) .toInt else 10
Val lines = spark.read.textFile (args (0)) .rdd
Val links = lines.map {s = >
Val parts = s.split ("\\ s +")
(parts (0), parts (1))
}. Distinct (). GroupByKey (). Cache ()
Var ranks = links.mapValues (v = > 1.0)
For (I)
Val size = urls.size
Urls.map (url = > (url, rank / size))
}
Ranks = contribs.reduceByKey (_ + _). MapValues (0.150.85 * _)
}
Val output = ranks.collect ()
Output.foreach (tup = > println (s "${tup._1} has rank: ${tup._2}."))
Spark.stop ()
}
} after reading the above, have you mastered how Spark implements PageRank? If you want to learn more skills or want to know more about it, you are welcome to follow the industry information channel, thank you for reading!
