Spark算子：RDD键值转换操作–groupByKey、reduceByKey、reduceByKeyLocally；groupByKey优化

偏执的太偏执、 2021-09-25 01:24 336阅读 0赞

关键字：Spark算子、Spark RDD键值转换、groupByKey、reduceByKey、reduceByKeyLocally

## groupByKey ##

def groupByKey(): RDD\[(K, Iterable\[V\])\] **//返回新RDD,value值类型是Iterable**

def groupByKey(numPartitions: Int): RDD\[(K, Iterable\[V\])\]

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[0] at makeRDD at <console>:24
    
    scala>  rdd1.groupByKey().collect
    res0: Array[(String, Iterable[Int])] = Array((A,CompactBuffer(0, 2)), (B,CompactBuffer(1, 2)), (C,CompactBuffer(1)))
    
    scala>  rdd1.groupByKey()
    res2: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[3] at groupByKey at <console>:26
    
    scala>

reduceByKey

def reduceByKey(func: (V, V) => V): RDD\[(K, V)\]**//返回新RDD,value值类型是原类型**

def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD\[(K, V)\]

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

# #

## reduceByKeyLocally ##

def reduceByKeyLocally(func: (V, V) => V): Map\[K, V\]

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

spark version:2.4.5,Scala version 2.11.12

转自：

[http://lxw1234.com/archives/tag/spark%E7%AE%97%E5%AD%90][http_lxw1234.com_archives_tag_spark_E7_AE_97_E5_AD_90]

*  如果可以用 `reduceByKey`,就用它，因为它会在map 端，先进行本地 combine ，可以大大减少要传输到 reduce 端的数据量，减少网络传输的开销；
 *  **只有在 reduceByKey 处理不了的时候，才用 `groupByKey().map()` 代替**

val counts = pairs.reduceByKey(_ +_)
    
    val counts = pairs.groupByKey().map(e => (e._1,e._2.sum))

[https://blog.csdn.net/u012292754/article/details/86291309][https_blog.csdn.net_u012292754_article_details_86291309]

[http_lxw1234.com_archives_tag_spark_E7_AE_97_E5_AD_90]: http://lxw1234.com/archives/tag/spark%E7%AE%97%E5%AD%90
[https_blog.csdn.net_u012292754_article_details_86291309]: https://blog.csdn.net/u012292754/article/details/86291309