Spark中实现基础的PageRank

吴军博士在《数学之美》中深入浅出地介绍了由Google的佩奇与布林提出的PageRank算法,这是一种民主表决式网页排名技术。书中提到PageRank的核心思想为:“在互联网上,如果一个网页被很多其他网页所链接,说明它受到普遍的承认和信赖,那么它的排名就高。”同时,该算法还要对来自不同网页的链接区别对待,排名越高的网页,则其权重会更高,即所谓网站贡献的链接权更大。

例如网页Y被X1,X2,X3,X4四个网页所链接,且这四个网页的权重分别为0.001,0.01,0.02,0.04,则网页Y的Rank值=0.01+0.02+0.03+0.04=0.071。但问题是,如何获得X1,X2,X3,X4这些网页的权重呢?答案是权重等于这些网页自身的Rank。然而,这些网页的Rank又是通过链接它的网页的权重计算而来,于是就陷入了“鸡与蛋”的怪圈。解决办法是为所有网页设定一个相同的Rank初始值,然后利用迭代的方式来逐步求解。

在《数学之美》第10章的延伸阅读中,有更详细的算法介绍,有兴趣的同学可以自行翻阅。下面是PageRank的简单执行步骤:

  1. 首先假定所有网页的初始Rank值为1/N,N为所有网页的数量。
  2. 开始迭代。每次迭代,则页面p会将r/n的值发送给所有链接了p页面的邻居页面。其中,r为当前页面的rank值,n为链接了当前页面的邻居页面数。该值实则就是当前页面p这次迭代的贡献者(contribution)。
  3. 每次迭代结束时,都对最终获得的contributions进行求和。假设每个contribution为c(i),则可以通过公式α/N + (1-α)∑c(i)获得每个页面的rank值。其中,α是一个常数值,可以认为是一个调优参数(tuning parameter),N为所有页面的数量。

究竟应该迭代多少次呢?由于PageRank实则是线性代数中的矩阵计算,佩奇和拉里已经证明了这个算法是收敛的。当两次迭代获得结果差异非常小,接近于0时,就可以停止迭代计算。《数学之美》中提到:“一般来讲,只要10次左右的迭代基本上就收敛了。”

我们将初始值进一步简化为1.0,并且将α/N的值设置为0.15,则公式α/N + (1-α)∑c(i)就变成0.15 + 0.85*∑c(i),那么在Spark中的实现为:

  val sc = new SparkContext(...)
  // 假定邻居页面的List存储为Spark objectFile
  val links = sc.objectFile[(String, Seq[String])]("links")
    .partitionBy(new HashPartitioner(100))
    .persist()

  //设置页面的初始rank值为1.0
  var ranks = links.mapValues(_ => 1.0)

  //迭代10次
  for (i <- 0 until 10) {
    val contributions = links.join(ranks).flatMap {
      case (pageId, (links, rank)) =>
        //注意此时的links为模式匹配获得的值,类型为Seq[String],并非前面读取出来的页面List
        links.map(dest => (dest, rank / links.size))
    }
    //简化了的rank计算公式
    ranks = contributions.reduceByKey(_ + _).mapValues(0.15 + 0.85 * _) 
  } 
  ranks.saveAsTextFile("ranks")

这段代码来自于Learning Spark: Lightning-fast big data analytics一书第4章。它充分地展现了Spark在进行数据分析的优雅与强大。虽然是简化了的PageRank算法,但如此精简的代码量仍然值得称赞。此外,该实现的性能比较Hadoop而言,也有显著提升。在Matei的论文An Architecture for Fast and General Data Processing on Large Clusters中,给出了这样的性能benchmark比较,如下图所示:

pagerank
图:Hadoop和Spark关于PageRank算法的性能比较(2014年)

注意,图中比较了Basic Spark与Spark+Controlled Partitioning,后者实则是通过分区等性能调优手段改进算法的性能。在前面的代码段中,我们也看到在读取页面List时,通过了partitionBy()函数传递了一个HashPartitioner。不要小看这段代码,实则它隐藏了许多与性能相关的tips,Learning Spark: Lightning-fast big data analytics一书对此做了深入介绍,性能改进点包括:

  • 在迭代中,links每次都与ranks进行了join操作,这是非常影响性能的。由于links的内容是不会变的(static dataset),因此在对它进行分区后,迭代中就不再需要针对它跨网络进行shuffle了。当links的数据量非常大时,这一优化对性能的提升是非常明显的。
  • 将分区了的RDD持久化到内存中,这一做法极为关键,理由同前。
  • 注意ranks的创建是针对links执行mapValues()而来。ranks需要和links执行join操作。由于mapValues()是一个transform操作,links是ranks的parent RDD,相比join两个完全无关的RDD而言,二者的join操作会更加高效。
  • 在迭代中,对contributions执行了reduceByKey()后,紧跟着执行了mapValues()。由于reduceByKey是hash-partition,它又是mapValues的parent RDD,因而mapValues()计算后的RDD即ranks也是hash-partition。而ranks又会在下一个迭代中与前面hash-partition的links进行join操作。join的两个RDD都处于同一分区,效率更高。

此外,对分区后的RDD应尽量调用mapValue()函数,而非map()函数。从Spark对mapValues的实现以及注释可以看到,mapValues()函数会保持原有RDD的分区:

  /**
   * Pass each value in the key-value pair RDD through a map function without changing the keys;
   * this also retains the original RDD's partitioning.
   */
  def mapValues[U](f: V => U): RDD[(K, U)] = {
    val cleanF = self.context.clean(f)
    new MappedValuesRDD(self, cleanF)
  }

flatMapValues()与mapValues()相似,也会保持原有RDD的分区,所以为了尽可能地满足与分区有关的性能优化,应合理考虑对RDD操作的选择。

2015-02-14 10:4555Spark