package com.lyzx.day37

import org.apache.spark.{Partitioner, SparkConf, SparkContext}

class D1 {

  //partitionBy和自定義分割槽器解決資料傾斜的問題
  def f1(sc:SparkContext): Unit ={
     val  r1 = sc.parallelize(1 to 100,4)
     val filterRdd = r1.filter(_ > 49)
      filterRdd.mapPartitionsWithIndex((idx,itr)=>{
        while(itr.hasNext){
          println("["+idx+"]"+itr.next())
        }
        for(v <- itr) yield  v
      }).collect()
    println(filterRdd.partitions.length)
    val reP = filterRdd
                .map(x=>(x,null))
                .partitionBy(new P(r1.partitions.length))
                  .map(x=>x._1)

    reP.mapPartitionsWithIndex((idx,itr)=>{
      while(itr.hasNext){
        println("["+idx+"]"+itr.next())
      }
      for(v <- itr) yield  v
    }).collect()
  }


  //只對K,V鍵值對的鍵做操作
  def f2(sc:SparkContext): Unit ={
    val  r1 = sc.parallelize(1 to 10).map(x=>(x,x))
    r1.mapValues(_+1)
      .foreach(println)
    println("====================================")

    //做完mapValues之後再壓平 可以把一個值對映為一個集合
    //在壓平的過程中把集合中的值掏出來,組合為(K1,V1),(K1,V2)的格式
    r1.flatMapValues(x=>Seq(x+100,x))
        .foreach(println)
  }


}

object D1{

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("day36").setMaster("local")
    val sc = new SparkContext(conf)
    val t = new D1

//    t.f1(sc)
    t.f2(sc)
    sc.stop()
  }
}

//自定義分割槽器,把資料均勻的分配到各個Partition上
class P(num:Int) extends  Partitioner{
  private var N:Int = 0
  override def numPartitions: Int = num

  //這麼做就和key沒有關係,只是完全按照資料的個數把他們分配到各個Partition中
  override def getPartition(key: Any): Int = {
    val currentIndex = (N % num)
    N += 1
    N = N % num  //這樣做是為了讓N保持在0 - num-1 之間防止N超出long的範圍
    currentIndex
  }
}