一 RDD Key -Value类型转换算子

1 groupByKey

函数签名
def groupByKey(): RDD[(K, Iterable[V])]
def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]
def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]
1
2
3
4

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

val rdd: RDD[(String,Int)] = sc.makeRDD(
  List(
    ("a", 1),
    ("a", 1),
    ("a", 1),
    ("b", 1)
  )
)

val rdd1: RDD[(String, Iterable[Int])] = rdd.groupByKey()

val rdd2: RDD[(String, Int)] = rdd1.mapValues(_.size)
1
2
3
4
5
6
7
8
9
10
11
12

groupByKey也可以实现WordCount（3 / 10）

（1）groupByKey 和 groupBy的区别

• groupBy不需要考虑数据类型，groupByKey 必须保证数据类型为KV类型
• groupBy按照指定的规则进行分组，groupByKey 必须根据K对V分组
• 返回结果类型
  • groupBy ==> (String, Iterable[(String, Int)])，按照规则分组，将KV放在一起
  • groupByKey ==> (String, Iterable[Int])，将V分在一个组中

groupBy 源码，底层使用的是groupByKey，但有以上几点区别

def groupBy[K](f: T => K, p: Partitioner)(implicit kt: ClassTag[K], ord: Ordering[K] = null)
    : RDD[(K, Iterable[T])] = withScope {
  val cleanF = sc.clean(f)
  this.map(t => (cleanF(t), t)).groupByKey(p)
}
1
2
3
4
5

（2）groupByKey 和 reduceByKey 的区别

• groupByKey ：将数据落盘，之后执行shuffle，按照key分组，最后使用一个新的RDD进行聚合

• reduceByKey：在落盘之前，每个分区内相同的key先进行聚合，称为预聚合（combine），之后再将剩余的数据落盘，再执行shuffle，按照key分区，直接在同一个RDD内进行聚合

  在落盘前将数据量减少，第二个RDD读取的数据也少了，shuffle阶段更快

  如果抛开combine阶段不谈，两者性能相差不多

从shuffle的角度：reduceByKey和groupByKey都存在shuffle的操作，但是reduceByKey可以在shuffle前对分区内相同key的数据进行预聚合（combine）功能，这样会减少落盘的数据量，而groupByKey只是进行分组，不存在数据量减少的问题，reduceByKey性能比较高

从功能的角度：reduceByKey其实包含分组和聚合的功能。groupByKey只能分组，不能聚合，所以在分组聚合的场合下，推荐使用reduceByKey，如果仅仅是分组而不需要聚合。那么还是只能使用groupByKey

2 aggregateByKey

函数签名
def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U,
  combOp: (U, U) => U): RDD[(K, U)]
1
2
3

将数据根据不同的规则进行分区内计算和分区间计算

需求：取出每个分区内相同key的最大值然后分区间相加

分析：

【（a，1），（a，2），（b，3）】=> 【（a，2），（b，3）】
													=> 【（a，8），（b，8）】
【（b，4），（b，5），（a，6）】=> 【（b，5），（a，6）】
1
2
3

groupByKey 在执行的时候，分区内和分区间计算逻辑相同，所以完不成此需求

aggregateByKey算子存在函数柯里化

• 第一个参数列表中有一个参数
  • 参数为零值（zeroValue: U），表示计算初始值，zero和z也同样是此含义
  • 零值的出现是因为在计算过程中，第一个K的V（第一个值）和第一次出现的K的V无法进行两两计算，使用零值与第一个V进行计算
  • 零值用于数据进行分区内计算
• 第二个参数列表有两个参数，其中
  • 第一个参数（seqOp: (U, V) => U）表示分区内计算规则
  • 第二个参数（combOp: (U, U) => U）表示分区间计算规则

以下代码中，第一个x为零值，y为各分区内K的V

第二个x为第一个分区挑选出的K的V，y为第二个分区内挑选出的V

零值为（a，0）（b，0）

val rdd: RDD[(String,Int)] = sc.makeRDD(
  List(
    ("a", 1),
    ("a", 2),
    ("b", 3),
    ("b", 4),
    ("b", 5),
    ("a", 6),
  ),2
)

val rdd1: RDD[(String, Int)] = rdd.aggregateByKey(0)(
  (x, y) => {
    Math.max(x, y)
  },
  (x, y) => {
    x + y
  }
)
rdd1.collect().foreach(println)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

在分区间存在shuffle，相同的key进入同一分区，进而聚合求和

aggregateByKey也可以实现WordCount（3 / 10）

val rdd2: RDD[(String, Int)] = rdd.aggregateByKey(0)(
  (x, y) => {
    x + y
  },
  (x, y) => {
    x + y
  }
)
1
2
3
4
5
6
7
8

简化

val rdd2: RDD[(String, Int)] = rdd.aggregateByKey(0)(_ + _,_ + _)
1

3 foldByKey

函数签名
def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]
1
2

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

rdd.foldByKey(0)(_ + _)
1

foldByKey也可以实现WordCount（5 / 10）

4 combineByKey

函数签名
def combineByKey[C](
  createCombiner: V => C,
  mergeValue: (C, V) => C,
  mergeCombiners: (C, C) => C): RDD[(K, C)]
1
2
3
4
5

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

需求：求每个key的平均值

分析：

("a", 1),("a", 2),("b", 3),
							=> （a,3），（b，4）
("b", 4),("b", 5),("a", 6),
1
2
3

reduceByKey可以求出总数，但不能求出count，avg = total / count

aggregateByKey(z)(f1,f2)强调的是分区内和分区间计算规则不同，同样count不容易得到

foldByKey(z)(f1)同样

groupByKey没有聚合能力

combineByKey的首要任务就是去进行数量的统计，此算子有三个参数

• 第一个createCombiner表示当第一个数据（首位数据，和第一个K不同的数据）格式不符合规则时，用于进行转换操作
• 第二个mergeValue表示分区内计算规则
• 第三个mergeCombiners表示分区间计算规则

（1）数据转换

如果将数据转换成以下格式，就可以进行求平均操作

("a", (1,1)),("a", (2,1)),("b", (3,1)),
("b", (4,1)),("b", (5,1)),("a", (6,1)),
1
2

但由于第一个参数的限制，最终数据变为，仍然可以进行求平均操作

("a", (1,1)),("a", 2),("b", (3,1)),
("b", (4,1)),("b", 5),("a", (6,1)),
1
2

val rdd: RDD[(String,Int)] = sc.makeRDD(
  List(
    ("a", 1),
    ("a", 2),
    ("b", 3),
    ("b", 4),
    ("b", 5),
    ("a", 6),
  ),2
)
val rdd1: RDD[(String, (Int, Int))] = rdd.combineByKey(
  num => (num, 1),
  (x: (Int, Int), y: Int) => {
    (x._1 + y, x._2 + 1)
  },
  (x: (Int, Int), y: (Int, Int)) => {
    (x._1 + y._1, x._2 + y._2)
  }
)

rdd1.collect().foreach(println)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

分区内，分区间计算结果为一个Tuple

combineByKey也可以完成WordCount（6 / 10）

val rdd1: RDD[(String, Int)] = rdd.combineByKey(
  num => num,
  (x: Int, y: Int) => {
    x + y
  },
  (x: Int, y: Int) => {
    (x + y)
  }
)
1
2
3
4
5
6
7
8
9

rdd.reduceByKey(_+_)
rdd.aggregateByKey(0)(_+_,_+_)
rdd.foldByKey(0)(_+_)
1
2
3

（2）四者的联系与区别-源码

reduceByKey

reduceByKey(func)
    
    combineByKeyWithClassTag[V](
    	(v: V) => v, 	//分区内第一个key的Value数据的转换
    	func, 		//分区内计算规则
    	func		//分区内计算规则
    )
1
2
3
4
5
6
7

aggregateByKey

aggregateByKey(zeroValue)(seqOp,combOp)
	
	combineByKeyWithClassTag[U](
		(v: V) => cleanedSeqOp(createZero(), v),	//分区内第一个key的Value数据的转换，初始													   值和v在做分区间计算
		cleanedSeqOp, 	//分区内计算规则
		combOp			//分区间计算规则
    )
1
2
3
4
5
6
7

foldByKey

foldByKey(zeroValue)(func)

	combineByKeyWithClassTag[V](
		(v: V) => cleanedFunc(createZero(), v),	//分区内第一个key的Value数据的转换，初始值和
												  v在做分区间计算
		cleanedFunc, 	//分区内计算规则
		cleanedFunc		//分区间计算规则
    )
1
2
3
4
5
6
7
8

combineByKey

combineByKey(createCombiner,mergeValue,mergeCombiners)

	combineByKeyWithClassTag(
		createCombiner, 	//分区内第一个key的Value数据的转换
		mergeValue, 		//分区内计算规则
		mergeCombiners)		//分区间计算规则
1
2
3
4
5
6

以上四者底层调用同一个方法，都有map端的预聚合功能（mapSideCombine: Boolean = true），而groupByKey没有

groupByKey

groupByKey()

	combineByKeyWithClassTag(
	createCombiner, 
	mergeValue, 
	mergeCombiners, 
	mapSideCombine = false)
1
2
3
4
5
6
7

5 sortByKey

函数签名
def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length)
  : RDD[(K, V)]
1
2
3

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

此算子按照K排序，默认升序

val rdd: RDD[(String,Int)] = sc.makeRDD(
  List(
    ("a", 1),
    ("c", 2),
    ("b", 3),
    ("c", 4),
    ("b", 5),
    ("a", 6),
    ("c", 2)
  )
)

val rdd1: RDD[(String, Int)] = rdd.sortByKey()
1
2
3
4
5
6
7
8
9
10
11
12
13

自定义K

val rdd: RDD[(User,Int)] = sc.makeRDD(
  List(
    (new User(), 1),
    (new User(), 2),
    (new User(), 3),
    (new User(), 4),
  )
)

val rdd1: RDD[(User, Int)] = rdd.sortByKey()

class User extends Ordered[User]{
    override def compare(that: User): Int = {
      1
    }
  }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

6 join

函数签名
def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
1
2

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

val rdd: RDD[(String,Int)] = sc.makeRDD(
  List(
    ("a", 1),
    ("b", 2),
    ("c", 3),
  )
)
val rdd1: RDD[(String,Int)] = sc.makeRDD(
  List(
    ("a", 4),
    ("d", 5),
    ("c", 6),
  )
)

val rdd2: RDD[(String, (Int, Int))] = rdd.join(rdd1)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

join操作可能产生笛卡尔乘积，可能会出现shuffle，性能比较差，所以如果能使用其他方式实现同样的功能，不推荐使用join

val rdd: RDD[(String,Int)] = sc.makeRDD(
  List(
    ("a", 1),
    ("a", 2),
    ("a", 3),
  )
)

val rdd1: RDD[(String,Int)] = sc.makeRDD(
  List(
    ("a", 4),
    ("a", 5),
    ("a", 6),
  )
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

7 left(right,full)OuterJoin

函数签名
def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
1
2

类似于SQL语句的左外连接，如果主表在join的左边就叫做左连接，主表在join的右边就叫右连接

主表：无论何时，数据全部呈现出来

从表：满足条件的数据才会呈现出来

val rdd: RDD[(String,Int)] = sc.makeRDD(
  List(
    ("a", 1),
    ("b", 2),
    ("c", 3),
  )
)
val rdd1: RDD[(String,Int)] = sc.makeRDD(
  List(
    ("a", 4),
    ("d", 5),
    ("c", 6),
  )
)
val rdd2: RDD[(String, (Int, Option[Int]))] = rdd.leftOuterJoin(rdd1)
val rdd3: RDD[(String, (Option[Int], Int))] = rdd.rightOuterJoin(rdd1)
val rdd4: RDD[(String, (Option[Int], Option[Int]))] = rdd.fullOuterJoin(rdd1)
rdd2.collect().foreach(println)
println("**********************")
rdd3.collect().foreach(println)
println("**********************")
rdd4.collect().foreach(println)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

Option只用两个对象，一个Some一个None

8 cogroup

函数签名
def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]
1
2

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

val rdd5: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd.cogroup(rdd1)
rdd5.collect().foreach(println)
1
2

cogroup，先进行组内分组，再连接，connect + group

最多可以将4个RDD组合成一个RDD

二 案例

数据准备—agent.log：时间戳，省份，城市，用户，广告，中间字段使用空格分隔

需求：统计出每一个省份每个广告被点击数量排行的Top3

1 思路一

先统计再分组

def main(args: Array[String]): Unit = {
  val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Req")
  val sc = new SparkContext(conf)

  //读取文件，获取原始数据
  val lines: RDD[String] = sc.textFile("data/agent.log")
  //将原始数据进行结构转换 line => ((省份，广告),1)
  val wordToOne: RDD[((String, String), Int)] = lines.map(
    line => {
      val datas = line.split(" ")
      ((datas(1), datas(4)), 1)
    }
  )
  //将转换结构后的数据进行统计，分组聚合 ((省份，广告),1) => ((省份，广告),sum)
  val wordToSum: RDD[((String, String), Int)] = wordToOne.reduceByKey(_ + _)
  //将统计结果进行结构转换，将省份独立出来 ((省份，广告),sum) => (省份，(广告,sum))
  val wordToTuple: RDD[(String, (String, Int))] = wordToSum.map {
    case ((prv, adv), sum) => {
      (prv, (adv, sum))
    }
  }
  //将数据按照省份分组 (省份,List[(广告1,sum1),(广告2,sum2),(广告3,sum3)])
  val groupRDD: RDD[(String, Iterable[(String, Int)])] = wordToTuple.groupByKey()
  //将分组后的数据根据点击数量进行排行，降序
  //将排序后的数据取前三
  val top3: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
    iter => {
      iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
    }
  )
  //将结果采集后打印在控制台上
  top3.collect().foreach(println)
  sc.stop()
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

2 思路二

先分组再统计

def main(args: Array[String]): Unit = {
  val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Req")
  val sc = new SparkContext(conf)

  //读取文件，获取原始数据
  val lines: RDD[String] = sc.textFile("data/agent.log")
  //将原始数据进行结构转换 line => (省份，(广告,1))
  val wordToOne: RDD[(String, (String, Int))] = lines.map(
    line => {
      val datas = line.split(" ")
      (datas(1), (datas(4), 1))
    }
  )
 val groupRDD: RDD[(String, Iterable[(String, Int)])] = wordToOne.groupByKey()

  val top3: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
    iter => {
      val wordCountMap: Map[String, Int] = iter.groupBy(_._1).mapValues(_.size)
      wordCountMap.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
    }
  )
  top3.collect().foreach(println)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

思路一核心逻辑为reduceByKey，思路二核心逻辑为groupByKey，一的效率更高，二会有大量的数据落盘操作

两者在内存中执行操作

一

val top3: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
  iter => {
    iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
  }
)
1
2
3
4
5

二

val top3: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
  iter => {
    val wordCountMap: Map[String, Int] = iter.groupBy(_._1).mapValues(_.size)
    wordCountMap.toList.sortBy(_._2)(Ordering.Int.reverse).take(3)
  }
)
1
2
3
4
5
6

思路二在内存中进行了大量数据的单点操作

所以先做统计分析，待数据量减少之后，再分组

3 优化

现有10G数据需要排序，但只有5M内存空间，如何完成：将10G数分划分为每5M一块，依次放入内存进行排序，之后再进行全局排序

从局部排序到全局排序

对以上需求进行分区内排序取前3名，分区间排序取前3名

def main(args: Array[String]): Unit = {
  val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Req")
  val sc = new SparkContext(conf)

  //读取文件，获取原始数据
  val lines: RDD[String] = sc.textFile("data/agent.log")
  //将原始数据进行结构转换 line => ((省份，广告),1)
  val wordToOne: RDD[((String, String), Int)] = lines.map(
    line => {
      val datas = line.split(" ")
      ((datas(1), datas(4)), 1)
    }
  )
  //将转换结构后的数据进行统计，分组聚合 ((省份，广告),1) => ((省份，广告),sum)
  val wordToSum: RDD[((String, String), Int)] = wordToOne.reduceByKey(_ + _)
  //将统计结果进行结构转换，将省份独立出来 ((省份，广告),sum) => (省份，(广告,sum))
  val wordToTuple: RDD[(String, (String, Int))] = wordToSum.map {
    case ((prv, adv), sum) => {
      (prv, (adv, sum))
    }
  }
  //[(广告1,sum1),(广告2,sum2),(广告3,sum3)]
  val top3: RDD[(String, ArrayBuffer[(String, Int)])] = wordToTuple.aggregateByKey(ArrayBuffer[(String, Int)]())(
    (buff, t) => {
      buff.append(t)
      buff.sortBy(_._2)(Ordering.Int.reverse).take(3)
    }, (
      (buff1, buff2) => {
        buff1.appendAll(buff2)
        buff1.sortBy(_._2)(Ordering.Int.reverse).take(3)
      }
      )
  )
  top3.collect().foreach(println)
  sc.stop()
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36