spark-shell如何實現PageRank

今天就跟大家聊聊有關spark-shell如何實現PageRank，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內容，希望大家根據這篇文章可以有所收獲。

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有進一步的了解嗎？如果還想了解更多知識或者相關內容，請關注億速云行業資訊頻道，感謝大家的支持。