package com.kun.core

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

import scala.collection.mutable.ArrayBuffer


/**
  * Dataset开发实战企业人员管理系统应用案例代码
  */
object DatasetOps {

  /**
    * Case Class的特别之处在于：
    *
    * 编译器会为Case Class自动生成以下方法：
    * equals & hashCode
    * toString
    * copy
    * 编译器会为Case Class自动生成伴生对象
    *
    * 编译器会为伴生对象自动生成以下方法
    * apply
    * unapply
    * 这意味着你可以不必使用new关键字来实例化一个case class.
    *
    * case class的类参数在不指定val/var修饰时，会自动编译为val，即对外只读，如果需要case class的字段外部可写，可以显式地指定var关键字！
    */
  case class Person(name: String, age: Long)
  case class Score(n: String, score: Long)

  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val spark = SparkSession.builder().appName("DatasetOps").master("local[4]").
      config("spark.sql.warehouse", "/home/ghost/IdeaProjects/SparkBook/spark-warehouse").getOrCreate()

    import org.apache.spark.sql.functions._
    import spark.implicits._

    /**
      * Dataset中的tranformation和Action操作，Action类型的操作有：
      * show collect first reduce take count等
      * 这些操作都会产生结果，也就是说会执行逻辑计算的过程
      */
    val personsDF = spark.read.json("data/peoplemanagedata/people.json")
    val personScoresDF = spark.read.json("data/peoplemanagedata/peopleScores.json")
    // 生成Dataset
    val personsDS = personsDF.as[Person]
    val personScoresDS = personScoresDF.as[Score]

    println("使用groupBy算子进行分组：")
    /**
      * +-------+---+-----+
      * |   name|age|count|
      * +-------+---+-----+
      * | Justin| 29|    1|
      * |   Andy| 30|    1|
      * |Michael| 16|    1|
      * |Michael| 46|    1|
      * | Justin| 19|    1|
      * +-------+---+-----+
      */
    // $"columnName"  Scala short hand for a named column
    val personsDSGrouped = personsDS.groupBy($"name", $"age").count()
    personsDSGrouped.show()

    println("使用agg算子concat内置函数将姓名、年龄连接在一起成为单个字符串列 ：")

    /**
      * +-------+---+-----------------+
      * |   name|age|concat(name, age)|
      * +-------+---+-----------------+
      * | Justin| 29|         Justin29|
      * |   Andy| 30|           Andy30|
      * |Michael| 16|        Michael16|
      * |Michael| 46|        Michael46|
      * | Justin| 19|         Justin19|
      * +-------+---+-----------------+
      */
    personsDS.groupBy($"name", $"age").agg(concat($"name", $"age")).show()

    println("使用col算子选择列 ：")

    /**
      * +------------+------------+
      * |          _1|          _2|
      * +------------+------------+
      * |[16,Michael]|[Michael,88]|
      * |   [30,Andy]|  [Andy,100]|
      * | [19,Justin]| [Justin,89]|
      * | [29,Justin]| [Justin,89]|
      * |[46,Michael]|[Michael,88]|
      * +------------+------------+
      */
    personsDS.joinWith(personScoresDS, personsDS.col("name")===personScoresDS.col("n")).show()

    println("使用sum、avg等函数计算年龄总和、平均年龄、最大年龄、最小年龄、唯一年龄计数、平均年龄、当前时间等数据 ：")

    /**
      * +-------+--------+--------+--------+--------+----------+-------------------+--------+--------------+
      * |   name|sum(age)|avg(age)|max(age)|min(age)|count(age)|count(DISTINCT age)|avg(age)|current_date()|
      * +-------+--------+--------+--------+--------+----------+-------------------+--------+--------------+
      * |Michael|      62|    31.0|      46|      16|         2|                  2|    31.0|    2019-02-26|
      * |   Andy|      30|    30.0|      30|      30|         1|                  1|    30.0|    2019-02-26|
      * | Justin|      48|    24.0|      29|      19|         2|                  2|    24.0|    2019-02-26|
      * +-------+--------+--------+--------+--------+----------+-------------------+--------+--------------+
      */
    personsDS.groupBy($"name").agg(sum($"age"), avg($"age"), max($"age"), min($"age"), count($"age"),
      countDistinct($"age"), mean($"age"), current_date()).show()

    println("函数collect_list、collect_set比较，collect_list函数结果中包含重复元素；collect_set函数结果中无重复元素：")

    /**
      * +-------+------------------+-----------------+
      * |   name|collect_list(name)|collect_set(name)|
      * +-------+------------------+-----------------+
      * |Michael|[Michael, Michael]|        [Michael]|
      * |   Andy|            [Andy]|           [Andy]|
      * | Justin|  [Justin, Justin]|         [Justin]|
      * +-------+------------------+-----------------+
      */
    personsDS.groupBy($"name").agg(collect_list($"name"), collect_set($"name")).show()

    println("使用sample算子进行随机采样：")

    /**
      * +---+-------+
      * |age|   name|
      * +---+-------+
      * | 16|Michael|
      * | 30|   Andy|
      * | 19| Justin|
      * | 46|Michael|
      * +---+-------+
      * 不保证刚好0.5
      */
    personsDS.sample(false, 0.5).show()

    println("使用randomSplit算子进行随机切分：")

    /**
      * +---+------+
      * |age|  name|
      * +---+------+
      * | 29|Justin|
      * +---+------+
      *
      * +---+-------+
      * |age|   name|
      * +---+-------+
      * | 16|Michael|
      * | 19| Justin|
      * | 30|   Andy|
      * | 46|Michael|
      * +---+-------+
      *
      * Array里面两个数为权重
      */
    personsDS.randomSplit(Array(10, 20)).foreach(dataset => dataset.show())

    println("使用select算子选择列：")

    /**
      * +-------+
      * |   name|
      * +-------+
      * |Michael|
      * |   Andy|
      * | Justin|
      * | Justin|
      * |Michael|
      * +-------+
      */
    personsDS.select("name").show()

    println("使用joinWith算子关联企业人员信息、企业人员分数评分信息：")

    /**
      * +------------+------------+
      * |          _1|          _2|
      * +------------+------------+
      * |[16,Michael]|[Michael,88]|
      * |   [30,Andy]|  [Andy,100]|
      * | [19,Justin]| [Justin,89]|
      * | [29,Justin]| [Justin,89]|
      * |[46,Michael]|[Michael,88]|
      * +------------+------------+
      */
    personsDS.joinWith(personScoresDS, $"name"===$"n").show()

    println("使用join算子关联企业人员信息、企业人员分数评分信息：")

    /**
      * +---+-------+-------+-----+
      * |age|   name|      n|score|
      * +---+-------+-------+-----+
      * | 16|Michael|Michael|   88|
      * | 30|   Andy|   Andy|  100|
      * | 19| Justin| Justin|   89|
      * | 29| Justin| Justin|   89|
      * | 46|Michael|Michael|   88|
      * +---+-------+-------+-----+
      */
    personsDS.join(personScoresDS, $"name"===$"n").show()

    println("使用sort算子对年龄进行降序排序：")

    /**
      * +---+-------+
      * |age|   name|
      * +---+-------+
      * | 46|Michael|
      * | 30|   Andy|
      * | 29| Justin|
      * | 19| Justin|
      * | 16|Michael|
      * +---+-------+
      */
    personsDS.sort($"age".desc).show()

    def myFlatMapFunction(myPerson: Person, myEncoder: Person): Dataset[Person] = {
      personsDS
    }

    /**
      * +-------+---+
      * |     _1| _2|
      * +-------+---+
      * |Michael| 46|
      * |   Andy|100|
      * | Justin| 49|
      * | Justin| 59|
      * |Michael| 76|
      * +-------+---+
      */
    personsDS.flatMap(persons => persons match {
      case Person(name, age) if name=="Andy" => List((name, age+70))
      case Person(name, age) => List((name, age+30))
    }).show()

    /**
      * +-------+----+
      * |     _1|  _2|
      * +-------+----+
      * |Michael|1016|
      * |   Andy|1030|
      * | Justin|1019|
      * | Justin|1029|
      * |Michael|1046|
      * +-------+----+
      */
    personsDS.mapPartitions{ persons =>
      val result = ArrayBuffer[(String, Long)]()
      while (persons.hasNext){
        val person = persons.next()
        result += ((person.name, person.age+1000))
      }
      result.iterator
    }.show()

    println("使用dropDuplicates算子统计企业人员管理系统姓名无重复员工的记录：")

    /**
      * +---+-------+
      * |age|   name|
      * +---+-------+
      * | 16|Michael|
      * | 30|   Andy|
      * | 19| Justin|
      * +---+-------+
      */
    personsDS.dropDuplicates("name").show()

    /**
      * +---+-------+
      * |age|   name|
      * +---+-------+
      * | 46|Michael|
      * | 30|   Andy|
      * | 19| Justin|
      * | 16|Michael|
      * | 29| Justin|
      * +---+-------+
      */
    personsDS.distinct().show()

    println("使用repartition算子设置分区：")

    /**
      * 原分区数：1
      */
    println("原分区数：" + personsDS.rdd.partitions.size)
    val repartitionDs = personsDS.repartition(4)

    /**
      * repartition设置分区数:4
      */
    println("repartition设置分区数:" + repartitionDs.rdd.partitions.size)

    println("使用coalesce算子设置分区：")

    /**
      * repartition(numPartitions:Int):RDD[T]和coalesce(numPartitions:Int，shuffle:Boolean=false):RDD[T]
      *
      * 他们两个都是RDD的分区进行重新划分，repartition只是coalesce接口中shuffle为true的简易实现，（假设RDD有N个分区，需要重新划分成M个分区）
      *
      * 1）、N<M。一般情况下N个分区有数据分布不均匀的状况，利用HashPartitioner函数将数据重新分区为M个，这时需要将shuffle设置为true。
      *
      * 2）如果N>M并且N和M相差不多，(假如N是1000，M是100)那么就可以将N个分区中的若干个分区合并成一个新的分区，最终合并为M个分区，这时可以将shuff设置为false，在shuffl为false的情况下，如果M>N时，coalesce为无效的，不进行shuffle过程，父RDD和子RDD之间是窄依赖关系。
      *
      * 3）如果N>M并且两者相差悬殊，这时如果将shuffle设置为false，父子ＲＤＤ是窄依赖关系，他们同处在一个Ｓｔａｇｅ中，就可能造成spark程序的并行度不够，从而影响性能，如果在M为1的时候，为了使coalesce之前的操作有更好的并行度，可以讲shuffle设置为true。
      *
      * 总之：如果shuff为false时，如果传入的参数大于现有的分区数目，RDD的分区数不变，也就是说不经过shuffle，是无法将RDDde分区数变多的。
      *
      * coalesce 合并
       */
    val coalesced: Dataset[Person] = repartitionDs.coalesce(2)

    /**
      * coalesce设置分区数：2
      */
    println("coalesce设置分区数：" + coalesced.rdd.partitions.size)

    /**
      * +---+-------+
      * |age|   name|
      * +---+-------+
      * | 30|   Andy|
      * | 19| Justin|
      * | 29| Justin|
      * | 16|Michael|
      * | 46|Michael|
      * +---+-------+
      */
    coalesced.show

    spark.stop()

  }
}