自定义分区及获取

1.自定义分区方式

/**
  * Created by chh on 2016/5/21.
  * 自定义分区
  */
class DomainNamePartitioner(numsParts: Int) extends Partitioner{
  override def numPartitions: Int = numsParts

  override def getPartition(key: Any): Int = {
    val domain=new java.net.URL(key.toString).getHost()
    val code=(domain.hashCode % numPartitions)
    if(code < 0){
      code + numPartitions  //确保非负值
    } else {
      code
    }
  }

  override  def equals(other: Any): Boolean = other match {
    //这个实例是DomainNamePartitioner的实例，并且numPartitions相同，返回true
    case dnp: DomainNamePartitioner =>
      dnp.numPartitions ==numPartitions
    case _=> false
  }
}

2.一个pageRank实例

步骤
① 将每个页面的排序值初始化为1.0
②在每次迭代中，向每个有直接链接的页面，发送一个值为rank(p)/numNeighbors(p)（出链数目） 的贡献量
③将每个页面的排序值设置为0.15+0.85*contributionsReceived

object Spark {
  /*
         #以下是url的内容：
         www.baidu.com www.hao123.com
         www.baidu.com www.2345.com
         www.baidu.com www.zhouyang.com
         www.hao123.com www.baidu.com
         www.hao123.com www.zhouyang.com
         www.zhouyang.com www.baidu.com
      */
  def main(args :Array[String]): Unit ={
    val conf =new SparkConf().setAppName("PageRank").setMaster("local")
    val sc=new SparkContext(conf)
    val inputs =sc.textFile("F:\\url.txt")
    //url,[urls]
    val links =inputs.map(x=>(x.split(" ")(0),x.split(" ")(1)))
      .distinct()
      .groupByKey()
      .cache()
    //url,rank
    var ranks =links.mapValues(value =>1.0)
    for(i<-0 until 10){

      val contribs =links.join(ranks).flatMap({
        case(pageid,(links,rank))=>
          //url Double
          links.map(dest=>(dest,rank/links.size))
      })
      //reduce and add the contribs
      ranks=contribs.reduceByKey((x,y)=>x+y).mapValues(v => 0.15+0.85*v)
    }
    ranks.collect().foreach(println)
  }
}

/*
结果：
(www.hao123.com,0.3685546839262259)
(www.baidu.com,0.761571325242544)
(www.2345.com,0.3685546839262259)
(www.zhouyang.com,0.5269013026650011)
*/

3.分区的设置与获取

    val conf =new SparkConf().setAppName("wordCount").setMaster("local")
    val sc=new SparkContext(conf)
    val pairs=sc.parallelize(List((1,1),(2,2),(3,3)))
    val partitioned =pairs.partitionBy(new HashPartitioner(2))
    println(partitioned.partitioner)