1.
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

//kv类型算子
//PartitionBy：指定分区器来对上游的数据进行分区
object _01PartitonByDemo {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local").setAppName("getRDD")
        val sc = new SparkContext(conf)

        //注意 RDD必须是对偶类型才会调出PartitonBy方法
        //因为RDD隐式装换成立PairRDD
        val rdd: RDD[(String, Int)] = sc.makeRDD(List(("1", 2), ("2", 3), ("4", 4), ("3", 0)), 2)
        //自己指定分区器 3个 生成3个分区
        val result: RDD[(String, Int)] = rdd.partitionBy(new HashPartitioner(3))
        result.collect().foreach(println)
        //可以生成文件查看数据所在分区
        result.saveAsTextFile("_01")
    }
}

//(3,0)
//(1,2)
//(4,4)
//(2,3)

2.
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
//按照key分组，组内进行规约运算
object _02ReduceBykeyDemo {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local").setAppName("getRDD")
        val sc = new SparkContext(conf)

        val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 2), ("a", 3), ("b", 4), ("c", 0)), 2)
        //按照指定参数传入一个函数
        val result: RDD[(String, Int)] = rdd.reduceByKey((x: Int, y: Int) => {
            x + y
        })
        //可以查看分区数
        val result1: RDD[(Int, (String, Int))] = result.mapPartitionsWithIndex((index, iter) => {
            iter.map((index, _))
        })
        result1.collect().foreach(println)

        //0,(b,4))
        //(1,(a,5))
        //(1,(c,0))
    }
}

3.    
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
//作用：通过key进行分组操作，默认使用Hashpartitioner
//相同key的数据分到一个组中，形成一个对偶元组，
//    元组中的第一个元素，就是Key
//    元组中的第二个元素，就是同一个Key的多个value对应的迭代器
object _03GroupBykeyDemo {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local").setAppName("getRDD")
        val sc = new SparkContext(conf)

        val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 2), ("a", 3), ("b", 4), ("c", 0)), 2)
        //可以自己指定分区数
        val result: RDD[(String, Iterable[Int])] = rdd.groupByKey(4)
        result.collect().foreach(println)
        println("*************************")

        //需求2：按照要处理的RDD的元素的第二个值进行自定义分组
        val result2: RDD[(Int, Iterable[(String, Int)])] = rdd.groupBy((e: (String, Int)) => e._2)
        result2.saveAsTextFile("_03")

        //(b,CompactBuffer(4))
        //(c,CompactBuffer(0))
        //*************************
        //还有相应的生成三个分区

    }
}

4.
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

//          第一个参数列表
//         *             默认值
//         *  第二个参数列表：
//         *          seqOp: 表示预聚合的函数，也就是分区内的运算逻辑
//         *          combOp：表示分区间的运算逻辑
object _04AggregateDemo {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local").setAppName("getRDD")
        val sc = new SparkContext(conf)

        val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 2), ("a", 3), ("b", 4), ("c", 0)), 2)
        /**
         * 逻辑就是先在分区内逻辑运算，在再分区间逻辑运算
         */
        val result: RDD[(String, Int)] = rdd.aggregateByKey(0)(
            (x, y) => math.max(x, y)
            ,
            (x, y) => x + y
        )
        result.collect().foreach(println)
        //(b,4)
        //(a,3)
        //(c,0)
    }
}

5.
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
//aggregate函数的特例
//就是分区内和分区间的逻辑一样
object _05FloDByKeyDemo {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local").setAppName("getRDD")
        val sc = new SparkContext(conf)

        val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 2), ("a", 3), ("b", 4), ("c", 0)), 2)
        //使用一样的逻辑
        val result: RDD[(String, Int)] = rdd.foldByKey(0)(
            (x, y) => x + y
        )
        result.collect().foreach(println)

        //(b,4)
        //(a,5)
        //(c,0)
    }
}