Spark KV类型算子案例详解一

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))
        /
        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)
    }
}