import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession

object Spark03 {

  //二次排序
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local[2]").appName("03").getOrCreate()
    import spark.implicits._
    Logger.getRootLogger.setLevel(Level.WARN)


    //sparkcore 計算方式
    fun1(spark.sparkContext)
    //sparksql 計算方式
    fun2(spark)

    spark.stop()
  }


  //使用sparkcore實現(xiàn)
  /*
  sparkcore實現(xiàn)思路:
  第一步: 自定義 key 實現(xiàn)scala.math.Ordered接口和Serializeable接口
  第二步：將要進(jìn)行二次排序的數(shù)據(jù)加載, 將數(shù)據(jù)映射為 <key，value> 格式的RDD
  第三步：使用sortByKey 基于自定義的key進(jìn)行二次排序
  第四步：去掉排序的key,只保留排序的結(jié)果
   */
  def fun1( sc:SparkContext ):Unit ={
    val data = List("hadoop  spark scala spark", "scala  mr", "spark  scala  mr  java" )

    sc.parallelize(data, 3)
      // 正則表達(dá)式  \\s 表示空白字符
      .flatMap( _.split("\\s+")  )
      .map( (_,1) )
      .reduceByKey( _+_ )
      .map( bean =>{
        ( MySortKey(bean._1, bean._2), bean )
      })
      .sortByKey(false,1)
      .map(_._2)
      .foreach( println )
  }


  //使用sparksql實現(xiàn)
  /**
    使用sql完成二次排序
    select  *   from   tablename   order  by  field1, field2 desc
    */
  def fun2( spark:SparkSession ):Unit ={
    import spark.implicits._
    val data = List("hadoop  spark scala spark", "scala  mr", "spark  scala  mr  java" )

    spark.sparkContext.parallelize(data,3)
      .flatMap( _.split("\\s+")  )
      .map( (_,1) )
      .reduceByKey( _+_ )
      .toDF("word","count")
      .createOrReplaceTempView("temptable")

    val sql = s"select word,count from temptable order by count desc,word "
    spark.sql(sql).show(false)
    
  }


  //Ordered[自定義數(shù)據(jù)類型]      用來自定義排序規(guī)則
  case class MySortKey(first:String, second:Int) extends Ordered[MySortKey]{
    //重寫比較方法
    // 首先 按照second字段 數(shù)字大小降序, 之后按照 first字段 字典升序
    override def compare(that: MySortKey): Int = {
      val temp = this.second - that.second
      if ( temp != 0 ){
        temp
      }else{
        - this.first.compareTo(that.first)
      }
    }
  }

}