Spark中的键值对操作-scala

val nums1 = sc.parallelize(List(Tuple2(1, 1), Tuple2(2, 2), Tuple2(1, 3),Tuple2(2, 4),Tuple2(3, 4)));
val nums2 = sc.parallelize(List(Tuple2(1,1),Tuple2(1,3),Tuple2(2,3)))
val results=nums1.cogroup(nums2)
for(tuple2 <- results.collect()){
  print(tuple2._1+" [ ")
  for(it <- tuple2._2._1)
    print(it+" ")
  print("] [ ")
  for(it<-tuple2._2._2)
    print(it+" ")
  println("]")
}

val nums =sc.parallelize(List(12,4,6,8,0,8));
//隐式转换声明排序的依据
implicit  val sortIntegersByString = new Ordering[Int] {
  override def compare(x: Int, y: Int): Int = x.toString().compareTo(y.toString())
}
val results=nums.sortBy(value=>value);
results.collect().foreach(println)

val nums = sc.parallelize(List(Tuple2(1, 1), Tuple2(2, 2), Tuple2(1, 3),Tuple2(2, 4),Tuple2(3, 4)));
//隐式转换声明排序的依据
implicit  val sortIntegersByString = new Ordering[Int] {
  override def compare(x: Int, y: Int): Int = x.toString().compareTo(y.toString())
}
val results=nums.sortByKey();
results.collect().foreach(println)

val list1 =List(Tuple2("zhou",List("it","math")),Tuple2("gan",List("money","book")))
val list2= List(Tuple2("zhou","it"),Tuple2("zhou","stock"),Tuple2("gan","money"),Tuple2("gan","book"))
val userData =sc.parallelize(list1)
val events = sc.parallelize(list2)
val joined=userData.join(events)
val results=joined.filter({
  case (id, (info, link)) =>
    !info.contains(link)
  }
).count()
println(results)

val list1 =List(Tuple2("zhou",List("it","math")),Tuple2("gan",List("money","book")))
val list2= List(Tuple2("zhou","it"),Tuple2("zhou","stock"),Tuple2("gan","money"),Tuple2("gan","book"))
val userData =sc.parallelize(list1).partitionBy(new HashPartitioner(100)).persist(StorageLevel.MEMORY_ONLY)

val list1 =List(Tuple2("zhou",List("it","math")),Tuple2("gan",List("money","book")))
val list2= List(Tuple2("zhou","it"),Tuple2("zhou","stock"),Tuple2("gan","money"),Tuple2("gan","book"))
val userData =sc.parallelize(list1).partitionBy(new HashPartitioner(100)).persist(StorageLevel.MEMORY_ONLY)
println(userData.partitioner)

/*
       #以下是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
    */
val inputs =sc.textFile("C:\\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)

class DomainNamePartitioner (numParts:Int) extends Partitioner{
  override def numPartitions: Int = numParts
  //根据hashCode和numPartitions取余来得到Partition,因为返回的必须是非负数，所以对于hashCode为负的情况做了特殊处理
  override def getPartition(key: Any): Int = {
    val domain = new 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
    //否则，返回false
    case _ => false
  }
}