Spark---RDD(Key-Value类型转换算子)

1.RDD Key-Value类型

Key-Value类型的算子即对键值对进行操作。

1.1 partitionBy

将数据按照指定的 Partitioner(分区器) 重新进行分区。Spark 默认的分区器为HashPartitioner，Spark除了默认的分区器外，常见的分区器还有：RangePartitioner、Custom Partitioner、SinglePartitioner等。

函数定义:
def partitionBy(partitioner: Partitioner): RDD[(K, V)]

	//使用HashPartitioner分区器并设置分区个数为2val data1: RDD[(Int, String)] = sparkRdd.makeRDD(Array((1, "aaa"), (2, "bbb"), (3, "ccc")), 3)data1.partitionBy(new HashPartitioner(2));data1.collect().foreach(println)

1.2 reduceByKey

可以将数据按照相同的 Key 对 Value 进行聚合

函数定义：
def reduceByKey(func: (V, V) => V): RDD[(K, V)]
def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]

	//将数据按照相同的key对value进行聚合val data1: RDD[(String, Int)] = sparkRdd.makeRDD(List(("a", 1), ("b", 2), ("c", 3),("a",4),("b",5)))val data2: RDD[(String, Int)] = data1.reduceByKey((x: Int, y: Int) => {x + y})data2.collect().foreach(println)

1.3 groupByKey

将数据源的数据根据 key 对 value 进行分组

函数定义：
def groupByKey(): RDD[(K, Iterable[V])]
def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]
def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]

    val dataRDD1 = sparkRdd.makeRDD(List(("a", 1), ("b", 2), ("c", 3),("a",4),("b",5)))val data1 = dataRDD1.groupByKey()//指定分区个数为2val data2 = dataRDD1.groupByKey(2)//指定分区器和分区个数val data3 = dataRDD1.groupByKey(new HashPartitioner(2))data1.collect().foreach(println)println("-------------------->")data2.collect().foreach(println)println("-------------------->")data3.collect().foreach(println)

reduceByKey和groupByKey的区别

从功能的角度来看：reduceByKey包含了分组和聚合功能，而groupByKey只包含了分组功能。
从shuffle的角度来看：为了避免占用过多的内存空间，reduceByKey和groupByKey在执行的过程中，都会执行shuffle操作，将数据打散写入到磁盘的临时文件中，而reduceByKey在进行shuffle前会对数据进行预聚合的操作，致使shuffle的效率得到的提升，因为减少了落盘的数据量。但是groupByKey在shuffle前不会进行预聚合操作。所以，reduceByKey在进行分组的时候，效率相对groupByKey来说较高。

reduceByKey:

groupByKey:

分区间和分区内

分区间： 顾名思义，分区间就是指的多个分区之间的操作。如reduceByKey在shuffle操作后将不同分区的数据传输在同一个分区中进行聚合。
分区内： 分区内字面意思指的是单个分区内之间的操作。如reduceByKey的预聚合功能就是在分区内完成

1.4 aggregateByKey

将数据根据不同的规则进行分区内计算和分区间计算，如reduceByKey中分区间和分区内都是聚合操作，而使用aggregateByKey可以设置分区间和分区内执行不同的操作。

函数定义：
def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U,
combOp: (U, U) => U): RDD[(K, U)]

    //取出每个分区内相同 key 的最大值然后分区间相加// aggregateByKey 算子是函数柯里化，存在两个参数列表// 1. 第一个参数列表中的参数表示初始值// 2. 第二个参数列表中含有两个参数// 2.1 第一个参数表示分区内的计算规则// 2.2 第二个参数表示分区间的计算规则val data1 = sparkRdd.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)),2)val data2 = data1.aggregateByKey(0)((x,y)=>{Math.max(x,y)},(x,y)=>{x+y})data2.collect().foreach(println)**

注意：最终的结果会受到设置的初始值的影响，返回结果的值的类型和初始值保持一致。

获取相同key的value的平均值

    val data1:RDD[(String,Int)] = sparkRdd.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4),("b",5),("a",6)),2)//设置初始值，初始值为一个元组，元组第一个元素表示value，第二个表示出现次数，初始默认都为0val data2:RDD[(String,(Int,Int))] = data1.aggregateByKey((0,0))((t, v)=> {(t._1 + v, t._2 + 1)} ,//分区内计算(t1, t2) => {(t1._1 + t2._1, t1._2 + t2._2)}//分区间计算)//和除以次数求出平均值val data3 = data2.mapValues({case (sum, count) => sum / count})data3.collect().foreach(println)

1.5 foldByKey

当分区内和分区间的计算规则相同的时候，aggregateByKey 就可以简化为 foldByKey

函数定义：
def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]

    val dataRDD1 = sparkRdd.makeRDD(List(("a",1),("b",2),("a",3)))val dataRDD2 = dataRDD1.foldByKey(0)(_+_)dataRDD2.collect().foreach(println)

1.6 combineByKey

最通用的对 key-value 型 rdd 进行聚集操作的聚集函数（aggregation function）。类似于aggregate()，combineByKey()允许用户返回值的类型与输入不一致。

函数定义：

def combineByKey[C](
createCombiner: V => C,//对数据进行转换
mergeValue: (C, V) => C, //分区内合并
mergeCombiners: (C, C) => C): RDD[(K, C)] //分区间合并

//将数据 List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98))求每个 key 的平均值
val rddSource: RDD[(String, Int)] = sparkRdd.makeRDD(List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)),2)
val combinRdd: RDD[(String, (Int, Int))] = rddSource.combineByKey(((x:Int)=>{(x,1)}),//对每个value进行转换，转换后为(value,1)，第一个元素为值，第二个元素为出现的次数((t1:(Int,Int),v)=>{(t1._1+v,t1._2+1)}),//分区内合并((t1,t2)=>{(t1._1+t2._1,t1._2+t2._2)})//分区间合并
)//mapValues算子是在key保持不变的时候对value进行操作val mapRdd: RDD[(String, Int)] = combinRdd.mapValues({case ((sum: Int, count: Int)) => sum / count})mapRdd.collect().foreach(println)

由此看出，combineByKey和aggreateByKey的不同之处在于，combineByKey可以不设置初始值，只需要对第一个元素进行转换，转换到合适的计算格式即可。

1.7 sortByKey

在一个(K,V)的 RDD 上调用，K 必须实现 Ordered 接口(特质)，返回一个按照 key 进行排序的

函数定义：

def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length)
: RDD[(K, V)]

//升序排序val dataRDD1 = sparkRdd.makeRDD(List(("a",1),("b",2),("c",3)))val sortRdd: RDD[(String, Int)] = dataRDD1.sortByKey()sortRdd.collect().foreach(print)

sortByKey默认为升序排序，如果想要降序排序，只需要将sortByKey第一个参数修改为false即可。

1.8 join

在类型为(K,V)和(K,W)的 RDD 上调用，返回一个相同 key 对应的所有元素连接在一起的(K,(V,W))的 RDD

函数定义：

def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]

	//join操作相当于数据库中的内连接，在连接的时候自动去除两边的悬浮元组val rdd0: RDD[(Int, String)] = sparkRdd.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))val rdd1: RDD[(Int, Int)] = sparkRdd.makeRDD(Array((1, 4), (2, 5), (3, 6)))rdd0.join(rdd1).collect().foreach(print)//修改rdd1，使其少了key=3的这个元素val rdd0: RDD[(Int, String)] = sparkRdd.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))val rdd1: RDD[(Int, Int)] = sparkRdd.makeRDD(Array((1, 4), (2, 5)))rdd0.join(rdd1).collect().foreach(print)

1.9 leftOuterJoin

类似于 SQL 语句的左外连接

函数定义：

def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]

    val rdd0: RDD[(Int, String)] = sparkRdd.makeRDD(List((1, "a"), (2, "b")))val rdd1: RDD[(Int, Int)] = sparkRdd.makeRDD(List((1, 4), (2, 5),(3, 6)))val rddRes = rdd0.leftOuterJoin(rdd1)rddRes.collect().foreach(print)

1.10 cogroup

在类型为(K,V)和(K,W)的 RDD 上调用，返回一个(K,(Iterable,Iterable))类型的 RDD，即先对

函数定义：

def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]

    val rdd0: RDD[(Int, String)] = sparkRdd.makeRDD(List((1, "a"), (2, "b"),(3,"c")))val rdd1: RDD[(Int, Int)] = sparkRdd.makeRDD(List((1, 4), (2, 5)))val rddRes = rdd0.cogroup(rdd1)rddRes.collect().foreach(print)