SparkCore算子之CombineByKey使用

Posted 2022-05-09 1610797071

SparkCore算子之CombineByKey使用

package com.bigdata.spark.core.rdd.oper.transform

import org.apache.spark.SparkConf, SparkContext

```scala
object RDD_Oper_Transform 

    def main(args: Array[String]): Unit = 
        val conf = new SparkConf().setMaster("local[*]").setAppName("Transform")
        val sc = new SparkContext(conf)
        val rdd = sc.makeRDD(
            List(
                ("a", 1), ("a", 2), ("b", 3),
                ("b", 4), ("b", 5), ("a", 6)
            ),
    2
        )
        // TODO 计算相同key的数据的平均值
        // 【 ("a", 1), ("a", 2), ("b", 3),】
        // 【 ("b", 4), ("b", 5), ("a", 6),】
        // (a, 3)(b, 4)
        // TODO reduceByKey算子更侧重于value的计算，不太关注数据的结构
        //rdd.reduceByKey
        // TODO aggregateByKey算子更侧重于分区内和分区间的计算逻辑不相同
        //rdd.aggregateByKey()
        // TODO foldByKey算子更侧重于分区内和分区间的计算逻辑相同
        //rdd.foldByKey()
        // TODO 转换数据结构
        // 【 ("a", 1), ("a", 2)】
        // 【 ("a", (1,1))), ("a", 2)】
        // 【 ("a", (3,2)), ("a", 6)】
        // 【 ("a", (9,3))】
//        rdd.map 
//            case ( word, cnt ) => 
//                (word, (cnt, 1))
//            
//        
//        .reduceByKey(
//            (t1, t2) => 
//                (t1._1 + t2._1, t1._2 + t2._2)
//            
//        )
//        .collect().foreach(println)
        // TODO combineByKey算子可以传递三个参数
        //   1. 第一个参数表示第一个key的数据的转换处理
        //   2. 第二个参数表示分区内计算规则
        //   3. 第三个参数表示分区间计算规则

//        rdd.combineByKey(
//            v => (v, 1),
//            (t : (Int, Int), v) => 
//                (t._1 + v, t._2 + 1)
//            ,
//            (t1:(Int, Int), t2:(Int, Int)) => 
//                (t1._1 + t2._1, t1._2 + t2._2)
//            
//        ).collect().foreach(println)
        // TODO combineByKey算子可以实现 WordCount ( 6 / 10 )
        //combineByKey 算子的计算过程中，数据结果的类型是执行后确定的。
        // 所以需要明确设定数据的类型，不能省略。
        rdd.combineByKey(
            v => v,
            (t :Int, v) => 
                t + v
            ,
            (t1:Int, t2:Int) => 
                t1 + t2
            
        ).collect().foreach(println)
        sc.stop()