Spark算子：transformation之键值转换groupByKey、reduceByKey、reduceByKeyLocally

待我称王封你为后i 2022-05-11 07:40 236阅读 0赞

1、**groupByKey**

1）def groupByKey(): RDD\[(K, Iterable\[V\])\]  
2）def groupByKey(numPartitions: Int): RDD\[(K, Iterable\[V\])\]  
3）def groupByKey(partitioner: Partitioner): RDD\[(K, Iterable\[V\])\]

该函数用于将RDD\[K,V\]中每个K对应的V值，合并到一个集合Iterable\[V\]中，参数numPartitions指分区数，partitioner指分区函数。

scala> var rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",1)))
    rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[89] at makeRDD at :21
     
    scala> rdd1.groupByKey().collect
    res81: Array[(String, Iterable[Int])] = Array((A,CompactBuffer(0, 2)), (B,CompactBuffer(2, 1)), (C,CompactBuffer(1)))

2、**reduceByKey**

1）def reduceByKey(func: (V, V) => V): RDD\[(K, V)\]  
2）def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD\[(K, V)\]  
3）def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD\[(K, V)\]

该函数将RDD\[K,V\]中的每个K对应的V值根据映射函数来计算。参数numPartitions指分区数，partitioner指分区函数。

scala> var rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",1)))
    rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[91] at makeRDD at :21
     
    scala> rdd1.partitions.size
    res82: Int = 15
     
    scala> var rdd2 = rdd1.reduceByKey((x,y) => x + y)
    rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[94] at reduceByKey at :23
     
    scala> rdd2.collect
    res85: Array[(String, Int)] = Array((A,2), (B,3), (C,1))
     
    scala> rdd2.partitions.size
    res86: Int = 15
     
    scala> var rdd2 = rdd1.reduceByKey(new org.apache.spark.HashPartitioner(2),(x,y) => x + y)
    rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[95] at reduceByKey at :23
     
    scala> rdd2.collect
    res87: Array[(String, Int)] = Array((B,3), (A,2), (C,1))
     
    scala> rdd2.partitions.size
    res88: Int = 2

val textRDD = sc.parallelize(List(("A", "aa"), ("B","bb"),("C","cc"),("C","cc"), ("D","dd"), ("D","dd")))
    val reducedRDD = textRDD.reduceByKey((value1,value2) => {value1+";"+value2})
    reducedRDD.collect.foreach(println)
    (D,dd;dd)
    (A,aa)
    (B,bb)
    (C,cc;cc)

3、**reduceByKeyLocally**：def reduceByKeyLocally(func: (V, V) => V): Map\[K, V\]

该函数将RDD\[K,V\]中每个K对应的V值根据映射函数来运算，运算结果映射到一个Map\[K,V\]中，而不是RDD\[K,V\]。