Spark Core快速入门系列(3) | ＜Transformation＞转换算子

2020-10-28 17:00:29 浏览数 (2)

在 RDD 上支持 2 种操作:

  • 1.transformation

  从一个已知的 RDD 中创建出来一个新的 RDD 例如: map就是一个transformation.

  • 2.action

  在数据集上计算结束之后, 给驱动程序返回一个值. 例如: reduce就是一个action.

  本篇博文可以学到 RDD 的转换操作, Action操作以后会详细讲解.

  在 Spark 中几乎所有的transformation操作都是懒执行的(lazy), 也就是说transformation操作并不会立即计算他们的结果, 而是记住了这个操作.

  只有当通过一个action来获取结果返回给驱动程序的时候这些转换操作才开始计算.这种设计可以使 Spark 运行起来更加的高效.默认情况下, 你每次在一个 RDD 上运行一个action的时候, 前面的每个transformed RDD 都会被重新计算.但是我们可以通过persist (or cache)方法来持久化一个 RDD 在内存中, 也可以持久化到磁盘上, 来加快访问速度. 后面有专门的章节学习这种持久化技术.

根据 RDD 中数据类型的不同, 整体分为 2 种 RDD:

  • 1.Value类型
  • 2.Key-Value类型(其实就是存一个二维的元组)

一. Value类型

1 map(func)

  • 1.作用:

  返回一个新的 RDD, 该 RDD 是由原 RDD 的每个元素经过函数转换后的值而组成. 就是对 RDD 中的数据做转换.

  • 2. 案例：创建一个包含1-10的的 RDD，然后将每个元素*2形成新的 RDD
代码语言：javascript复制
scala > val rdd1 = sc.parallelize(1 to 10)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24
// 得到一个新的 RDD, 但是这个 RDD 中的元素并不是立即计算出来的
scala> val rdd2 = rdd1.map(_ * 2)
rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[1] at map at
<console>:26

// 开始计算 rdd2 中的元素, 并把计算后的结果传递给驱动程序
scala> rdd2.collect
res0: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18, 20)

2 mapPartitions(func)

  • 1. 作用

  类似于map(func), 但是是独立在每个分区上运行.所以:Iterator<T> => Iterator<U>

  假设有N个元素，有M个分区，那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区。

  • 2. 案例：创建一个RDD，使每个元素*2组成新的RDD
代码语言：javascript复制
// 1. 创建一个RDD
scala> val source = sc.parallelize(1 to 10)
source: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at parallelize at <console>:24

// 2. 使每个元素*2组成新的RDD
scala> source.mapPartitions(it => it.map(_ * 2))
res7: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at mapPartitions at <console>:27

// 3. 打印新的RDD
scala> res7.collect
res8: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18, 20)

3 mapPartitionsWithIndex(func)

  • 1. 作用

  和mapPartitions(func)类似. 但是会给func多提供一个Int值来表示分区的索引. 所以func的类型是:(Int, Iterator<T>) => Iterator<U>

  • 2. 案例：创建一个RDD，使每个元素跟所在分区形成一个元组组成一个新的RDD
代码语言：javascript复制
// 1.创建一个RDD
scala> val rdd = sc.parallelize(Array(1,2,3,4))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:24

// 2.使每个元素跟所在分区形成一个元组组成一个新的RDD
scala> val indexRdd = rdd.mapPartitionsWithIndex((index,items)=>(items.map((index,_))))
indexRdd: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[5] at mapPartitionsWithIndex at <console>:26

// 3.打印新的RDD
scala> indexRdd.collect
res2: Array[(Int, Int)] = Array((0,1), (0,2), (1,3), (1,4))
  • 3. 分区数的确定, 和对数组中的元素如何进行分区
代码语言：javascript复制
// 1.确定分区数:
override def defaultParallelism(): Int = scheduler.conf.getInt("spark.default.parallelism", totalCores)

// 2.对元素进行分区
// length: RDD 中数据的长度  numSlices: 分区数
def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {
 (0 until numSlices).iterator.map { i =>val start = ((i * length) / numSlices).toInt
   val end = (((i   1) * length) / numSlices).toInt
   (start, end)
 }
}
seq match {
 case r: Range =>
   
 case nr: NumericRange[_] =>
   
 case _ =>
   val array = seq.toArray // To prevent O(n^2) operations for List etc
   positions(array.length, numSlices).map { case (start, end) =>
       array.slice(start, end).toSeq
   }.toSeq
}

4 map()和mapPartitions()的区别

  • map()：每次处理一条数据。
  • mapPartitions()：每次处理一个分区的数据，这个分区的数据处理完后，原 RDD 中该分区的数据才能释放，可能导致 OOM。
  • 开发指导：当内存空间较大的时候建议使用mapPartitions()，以提高处理效率。

5 flatMap(func)

  • 1. 作用

  类似于map，但是每一个输入元素可以被映射为 0 或多个输出元素（所以func应该返回一个序列，而不是单一元素 T => TraversableOnce[U]）

  • 2. 案例1：创建一个元素为1-5的RDD，运用flatMap创建一个新的RDD，新的RDD为原RDD的每个元素的2倍（2，4，6，8，10）
代码语言：javascript复制
// 1.创建
scala> val sourceFlat = sc.parallelize(1 to 5)
sourceFlat: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[12] at parallelize at <console>:24
// 2.打印
scala> sourceFlat.collect()
res11: Array[Int] = Array(1, 2, 3, 4, 5)
// 3.根据原RDD创建新RDD（1->1,2->1,2……5->1,2,3,4,5）
scala> val flatMap = sourceFlat.flatMap(1 to _)
flatMap: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[13] at flatMap at <console>:26
// 4.打印新RDD
scala> flatMap.collect()
res12: Array[Int] = Array(1, 1, 2, 1, 2, 3, 1, 2, 3, 4, 1, 2, 3, 4, 5)
  • 3. 案例2：创建一个元素为 1-5 的RDD，运用 flatMap创建一个新的 RDD，新的 RDD 为原 RDD 每个元素的 平方和三次方 来组成 1,1,4,8,9,27…
代码语言：javascript复制
scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at parallelize at <console>:24

scala> rdd1.flatMap(x => Array(x * x, x * x * x))
res13: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at flatMap at <console>:27

scala> res13.collect
res14: Array[Int] = Array(1, 1, 4, 8, 9, 27, 16, 64, 25, 125)

6 glom()

  • 1. 作用

  将每一个分区的元素合并成一个数组，形成新的 RDD 类型是RDD[Array[T]]

  • 2. 案例：创建一个 4 个分区的 RDD，并将每个分区的数据放到一个数组
代码语言：javascript复制
// 1.创建
scala> var rdd1 = sc.parallelize(Array(10,20,30,40,50,60), 4)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

// 2.将每个分区的数据放到一个数组并收集到Driver端打印
scala> rdd1.glom.collect
res2: Array[Array[Int]] = Array(Array(10), Array(20, 30), Array(40), Array(50, 60))

7 groupBy(func)

  • 1. 作用

  按照func的返回值进行分组.   func返回值作为 key, 对应的值放入一个迭代器中. 返回的 RDD: RDD[(K, Iterable[T])   每组内元素的顺序不能保证, 并且甚至每次调用得到的顺序也有可能不同.

  • 2. 案例1：创建一个RDD，按照元素模以2的值进行分组。
代码语言：javascript复制
// 1.创建
scala> val rdd = sc.parallelize(1 to 4)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[65] at parallelize at <console>:24
// 2.按照元素模以2的值进行分组
scala> val group = rdd.groupBy(_%2)
group: org.apache.spark.rdd.RDD[(Int, Iterable[Int])] = ShuffledRDD[2] at groupBy at <console>:26
// 3.打印结果
scala> group.collect
res0: Array[(Int, Iterable[Int])] = Array((0,CompactBuffer(2, 4)), (1,CompactBuffer(1, 3)))
  • 3. 案例2：创建一个 RDD，按照元素的奇偶性进行分组
代码语言：javascript复制
scala> val rdd1 = sc.makeRDD(Array(1, 3, 4, 20, 4, 5, 8))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[2] at makeRDD at <console>:24
    
scala> rdd1.groupBy(x => if(x % 2 == 1) "odd" else "even")
res4: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[4] at groupBy at <console>:27

scala> res4.collect
res5: Array[(String, Iterable[Int])] = Array((even,CompactBuffer(4, 20, 4, 8)), (odd,CompactBuffer(1, 3, 5)))

8 filter(func)

  • 1. 作用

  过滤。返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成。

  • 2. 案例：创建一个RDD，按照元素模以2的值进行分组。
代码语言：javascript复制
// 1.创建
scala> var sourceFilter = sc.parallelize(Array("xiaoming","xiaojiang","xiaohe","dazhi"))
sourceFilter: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[10] at parallelize at <console>:24
// 2.打印
scala> sourceFilter.collect()
res9: Array[String] = Array(xiaoming, xiaojiang, xiaohe, dazhi)
// 3.过滤出含” xiao”子串的形成一个新的RDD
scala> val filter = sourceFilter.filter(_.contains("xiao"))
filter: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[11] at filter at <console>:26
// 4.打印新RDD
scala> filter.collect()
res10: Array[String] = Array(xiaoming, xiaojiang, xiaohe)

9 sample(withReplacement, fraction, seed)

  • 1. 作用

  1.以指定的随机种子随机抽样出比例为fraction的数据，(抽取到的数量是: size * fraction). 需要注意的是得到的结果并不能保证准确的比例.

  2.withReplacement表示是抽出的数据是否放回，true为有放回的抽样，false为无放回的抽样. 放回表示数据有可能会被重复抽取到, false 则不可能重复抽取到. 如果是false, 则fraction必须是:[0,1], 是 true 则大于等于0就可以了.

  3.seed用于指定随机数生成器种子。 一般用默认的, 或者传入当前的时间戳

  • 2. 案例：创建一个RDD（1-10），从中选择放回和不放回抽样
代码语言：javascript复制
// 1.创建RDD
scala> val rdd = sc.parallelize(1 to 10)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[20] at parallelize at <console>:24

// 2.打印
scala> rdd.collect()
res15: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

// 3.放回抽样
scala> var sample1 = rdd.sample(true,0.4,2)
sample1: org.apache.spark.rdd.RDD[Int] = PartitionwiseSampledRDD[21] at sample at <console>:26

// 4.打印放回抽样结果
scala> sample1.collect()
res16: Array[Int] = Array(1, 2, 2, 7, 7, 8, 9)

// 5.不放回抽样
scala> var sample2 = rdd.sample(false,0.2,3)
sample2: org.apache.spark.rdd.RDD[Int] = PartitionwiseSampledRDD[22] at sample at <console>:26

// 6.打印不放回抽样结果
scala> sample2.collect()
res17: Array[Int] = Array(1, 9)

10 distinct([numTasks]))

  • 1. 作用

  对 RDD 中元素执行去重操作. 参数表示任务的数量.默认值和分区数保持一致.

  • 2. 案例：创建一个RDD，使用distinct()对其去重。
代码语言：javascript复制
// 1.创建一个RDD
scala> val distinctRdd = sc.parallelize(List(1,2,1,5,2,9,6,1))
distinctRdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[34] at parallelize at <console>:24

// 2.对RDD进行去重（不指定并行度）
scala> val unionRDD = distinctRdd.distinct()
unionRDD: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[37] at distinct at <console>:26

// 3.打印去重后生成的新RDD
scala> unionRDD.collect()
res20: Array[Int] = Array(1, 9, 5, 6, 2)

// 4.对RDD（指定并行度为2）
scala> val unionRDD = distinctRdd.distinct(2)
unionRDD: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[40] at distinct at <console>:26

// 5.打印去重后生成的新RDD
scala> unionRDD.collect()
res21: Array[Int] = Array(6, 2, 1, 9, 5)

11 coalesce(numPartitions)

  • 1. 作用

  缩减分区数，用于大数据集过滤后，提高小数据集的执行效率。

  • 2. 案例：创建一个4个分区的RDD，对其缩减分区
代码语言：javascript复制
// 1.创建一个RDD
scala> val rdd = sc.parallelize(1 to 16,4)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[54] at parallelize at <console>:24

// 2.查看RDD的分区数
scala> rdd.partitions.size
res20: Int = 4

// 3.对RDD重新分区
scala> val coalesceRDD = rdd.coalesce(3)
coalesceRDD: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[55] at coalesce at <console>:26

// 4.查看新RDD的分区数
scala> coalesceRDD.partitions.size
res21: Int = 3

注意:   第二个参数表示是否shuffle, 如果不传或者传入的为false, 则表示不进行shuffer, 此时分区数减少有效, 增加分区数无效.

12 repartition(numPartitions)

  • 1. 作用

  根据新的分区数, 重新 shuffle 所有的数据, 这个操作总会通过网络.新的分区数相比以前可以多, 也可以少

  • 2. 案例：创建一个4个分区的RDD，对其重新分区
代码语言：javascript复制
// 1.创建一个RDD
scala> val rdd = sc.parallelize(1 to 16,4)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[56] at parallelize at <console>:24

// 2.查看RDD的分区数
scala> rdd.partitions.size
res22: Int = 4

// 3.对RDD重新分区
scala> val rerdd = rdd.repartition(2)
rerdd: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[60] at repartition at <console>:26

// 4.查看新RDD的分区数
scala> rerdd.partitions.size
res23: Int = 2

13 coalasce和repartition的区别

  • coalesce重新分区，可以选择是否进行shuffle过程。由参数shuffle: Boolean = false/true决定。
  • repartition实际上是调用的coalesce，进行shuffle。源码如下：
代码语言：javascript复制
def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
coalesce(numPartitions, shuffle = true)
}
  • 如果是减少分区, 尽量避免 shuffle

14 sortBy(func,[ascending], [numTasks])

  • 1. 作用

  使用func先对数据进行处理，按照处理后结果排序，默认为正序。

  • 2. 案例1：创建一个RDD，按照不同的规则进行排序
代码语言：javascript复制
// 1.创建一个RDD
scala> val rdd = sc.parallelize(List(2,1,3,4))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[21] at parallelize at <console>:24

// 2.按照自身大小排序
scala> rdd.sortBy(x => x).collect()
res11: Array[Int] = Array(1, 2, 3, 4)

// 3.按照与3余数的大小排序
scala> rdd.sortBy(x => x%3).collect()
res12: Array[Int] = Array(3, 4, 1, 2)
  • 3. 案例2：
代码语言：javascript复制
scala> val rdd1 = sc.parallelize(Array(1,3,4,10,4,6,9,20,30,16))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[46] at parallelize at <console>:24

scala> rdd1.sortBy(x => x).collect
res17: Array[Int] = Array(1, 3, 4, 4, 6, 9, 10, 16, 20, 30)

scala> rdd1.sortBy(x => x, true).collect
res18: Array[Int] = Array(1, 3, 4, 4, 6, 9, 10, 16, 20, 30)

// 不用正序
scala> rdd1.sortBy(x => x, false).collect
res19: Array[Int] = Array(30, 20, 16, 10, 9, 6, 4, 4, 3, 1)

15 pipe(command, [envVars])

  • 1. 作用

  管道，针对每个分区，把 RDD 中的每个数据通过管道传递给shell命令或脚本，返回输出的RDD。一个分区执行一次这个命令. 如果只有一个分区, 则执行一次命令.

注意: 脚本要放在 worker 节点可以访问到的位置

  • 2. 案例：编写一个脚本，使用管道将脚本作用于RDD上。
代码语言：javascript复制
// 1: 创建一个脚本文件pipe.sh
文件内容如下:
echo "hello"
while read line;do
    echo ">>>"$line
done

// 2: 创建只有 1 个分区的RDD
scala> val rdd1 = sc.parallelize(Array(10,20,30,40), 1)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> rdd1.pipe("./pipe.sh").collect
res1: Array[String] = Array(hello, >>>10, >>>20, >>>30, >>>40)

// 3: 创建有 2 个分区的 RDD
scala> val rdd1 = sc.parallelize(Array(10,20,30,40), 2)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[3] at parallelize at <console>:24

scala> rdd1.pipe("./pipe.sh").collect
res2: Array[String] = Array(hello, >>>10, >>>20, hello, >>>30, >>>40)

总结: 每个分区执行一次脚本, 但是每个元素算是标准输入中的一行

二. 双 Value 类型交互

  这里的“双 Value 类型交互”是指的两个 RDD[V] 进行交互.

1. union(otherDataset)

  • 1. 作用

  求并集. 对源 RDD 和参数 RDD 求并集后返回一个新的 RDD

注意: union和 是等价的

  • 2. 案例：编写一个脚本，使用管道将脚本作用于RDD上。
代码语言：javascript复制
// 1.创建第一个RDD
scala> val rdd1 = sc.parallelize(1 to 5)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[23] at parallelize at <console>:24

// 2.创建第二个RDD
scala> val rdd2 = sc.parallelize(5 to 10)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[24] at parallelize at <console>:24

// 3.计算两个RDD的并集
scala> val rdd3 = rdd1.union(rdd2)
rdd3: org.apache.spark.rdd.RDD[Int] = UnionRDD[25] at union at <console>:28

// 4.打印并集结果
scala> rdd3.collect()
res18: Array[Int] = Array(1, 2, 3, 4, 5, 5, 6, 7, 8, 9, 10)

2. subtract (otherDataset)

  • 1. 作用

  计算差集. 从原 RDD 中减去 原 RDD 和 otherDataset 中的共同的部分.

  • 2. 案例：创建两个RDD，求第一个RDD与第二个RDD的差集
代码语言：javascript复制
// 1.创建第一个RDD
scala> val rdd = sc.parallelize(3 to 8)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[70] at parallelize at <console>:24
// 2.创建第二个RDD
scala> val rdd1 = sc.parallelize(1 to 5)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[71] at parallelize at <console>:24
// 3.计算第一个RDD与第二个RDD的差集并打印
scala> rdd.subtract(rdd1).collect()
res27: Array[Int] = Array(8, 6, 7)

3. intersection(otherDataset)

  • 1. 作用

  对源RDD和参数RDD求交集后返回一个新的RDD

  • 2. 案例：创建两个RDD，求两个RDD的交集
代码语言：javascript复制
// 1.创建第一个RDD
scala> val rdd1 = sc.parallelize(1 to 7)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[26] at parallelize at <console>:24

// 2.创建第二个RDD
scala> val rdd2 = sc.parallelize(5 to 10)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[27] at parallelize at <console>:24

// 3.计算两个RDD的交集
scala> val rdd3 = rdd1.intersection(rdd2)
rdd3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[33] at intersection at <console>:28

// 4.打印计算结果
scala> rdd3.collect()
res19: Array[Int] = Array(5, 6, 7)

4. cartesian(otherDataset)

  • 1. 作用

  计算 2 个 RDD 的笛卡尔积. 尽量避免使用

  • 2. 案例：创建两个RDD，计算两个RDD的笛卡尔积
代码语言：javascript复制
// 1.创建第一个RDD
scala> val rdd1 = sc.parallelize(1 to 3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[47] at parallelize at <console>:24

// 2.创建第二个RDD
scala> val rdd2 = sc.parallelize(2 to 5)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[48] at parallelize at <console>:24

// 3.计算两个RDD的笛卡尔积并打印
scala> rdd1.cartesian(rdd2).collect()
res17: Array[(Int, Int)] = Array((1,2), (1,3), (1,4), (1,5), (2,2), (2,3), (2,4), (2,5), (3,2), (3,3), (3,4), (3,5))

5. zip(otherDataset)

  • 1. 作用

  拉链操作. 需要注意的是, 在 Spark 中, 两个 RDD 的元素的数量和分区数都必须相同, 否则会抛出异常.(在 scala 中, 两个集合的长度可以不同)

类似算子: zipWithIndex, zipPartitions

  • 2. 案例：创建两个RDD，并将两个RDD组合到一起形成一个(k,v)RDD
代码语言：javascript复制
// 1.创建第一个RDD
scala> val rdd1 = sc.parallelize(Array(1,2,3),3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:24

// 2.创建第二个RDD（与1分区数相同）
scala> val rdd2 = sc.parallelize(Array("a","b","c"),3)
rdd2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[2] at parallelize at <console>:24

// 3.第一个RDD组合第二个RDD并打印
scala> rdd1.zip(rdd2).collect
res1: Array[(Int, String)] = Array((1,a), (2,b), (3,c))

// 4.第二个RDD组合第一个RDD并打印
scala> rdd2.zip(rdd1).collect
res2: Array[(String, Int)] = Array((a,1), (b,2), (c,3))

// 5.创建第三个RDD（与1,2分区数不同）
scala> val rdd3 = sc.parallelize(Array("a","b","c"),2)
rdd3: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[5] at parallelize at <console>:24

// 6.第一个RDD组合第三个RDD并打印
scala> rdd1.zip(rdd3).collect
java.lang.IllegalArgumentException: Can't zip RDDs with unequal numbers of partitions: List(3, 2)
  at org.apache.spark.rdd.ZippedPartitionsBaseRDD.getPartitions(ZippedPartitionsRDD.scala:57)
  at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:252)
  at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:250)
  at scala.Option.getOrElse(Option.scala:121)
  at org.apache.spark.rdd.RDD.partitions(RDD.scala:250)
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:1965)
  at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:936)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
  at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)
  at org.apache.spark.rdd.RDD.collect(RDD.scala:935)
  ... 48 elided

三. Key-Value 类型

  大多数的 Spark 操作可以用在任意类型的 RDD 上, 但是有一些比较特殊的操作只能用在key-value类型的 RDD 上.这些特殊操作大多都涉及到 shuffle 操作, 比如: 按照 key 分组(group), 聚集(aggregate)等.   在 Spark 中, 这些操作在包含对偶类型(Tuple2)的 RDD 上自动可用(通过隐式转换).

代码语言：javascript复制
object RDD {
  implicit def rddToPairRDDFunctions[K, V](rdd: RDD[(K, V)])
    (implicit kt: ClassTag[K], vt: ClassTag[V], ord: Ordering[K] = null): PairRDDFunctions[K, V] = {
    new PairRDDFunctions(rdd)
  }

  键值对的操作是定义在PairRDDFunctions类上, 这个类是对RDD[(K, V)]的装饰.

1. partitionBy

  • 1. 作用

  对pairRDD进行分区操作，如果原有的partionRDD和现有的partionRDD是一致的话就不进行分区， 否则会生成ShuffleRDD，即会产生shuffle过程。

  • 2. partitionBy 算子源码
代码语言：javascript复制
def partitionBy(partitioner: Partitioner): RDD[(K, V)] = self.withScope {
  
  if (self.partitioner == Some(partitioner)) {
    self
  } else {
    new ShuffledRDD[K, V, V](self, partitioner)
  }
}
  • 3. 案例1：创建一个4个分区的RDD，对其重新分区
代码语言：javascript复制
// 1.创建一个RDD
scala> val rdd = sc.parallelize(Array((1,"aaa"),(2,"bbb"),(3,"ccc"),(4,"ddd")),4)
rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[44] at parallelize at <console>:24

// 2.查看RDD的分区数
scala> rdd.partitions.size
res24: Int = 4

// 3.对RDD重新分区
scala> var rdd2 = rdd.partitionBy(new org.apache.spark.HashPartitioner(2))
rdd2: org.apache.spark.rdd.RDD[(Int, String)] = ShuffledRDD[45] at partitionBy at <console>:26

// 4.查看新RDD的分区数
scala> rdd2.partitions.size
res25: Int = 2
  • 4. 案例2：
代码语言：javascript复制
scala> val rdd1 = sc.parallelize(Array((1, "a"), (2, "b"), (3, "c"), (4, "d")))
rdd1: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> rdd1.partitions.length
res1: Int = 2

scala> rdd1.partitionBy(new org.apache.spark.HashPartitioner(3)).partitions.length
res3: Int = 3

2. groupByKey

  • 1. 作用

  groupByKey也是对每个key进行操作，但只生成一个sequence。

  • 2. 案例：创建一个pairRDD，将相同key对应值聚合到一个sequence中，并计算相同key对应值的相加结果。
代码语言：javascript复制
// 1.创建一个pairRDD
scala> val words = Array("one", "two", "two", "three", "three", "three")
words: Array[String] = Array(one, two, two, three, three, three)

scala> val wordPairsRDD = sc.parallelize(words).map(word => (word, 1))
wordPairsRDD: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[4] at map at <console>:26

// 2.将相同key对应值聚合到一个sequence中
scala> val group = wordPairsRDD.groupByKey()
group: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[5] at groupByKey at <console>:28

// 3.打印结果
scala> group.collect()
res1: Array[(String, Iterable[Int])] = Array((two,CompactBuffer(1, 1)), (one,CompactBuffer(1)), (three,CompactBuffer(1, 1, 1)))

// 4.计算相同key对应值的相加结果
scala> group.map(t => (t._1, t._2.sum))
res2: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[6] at map at <console>:31

// 5.打印结果
scala> res2.collect()
res3: Array[(String, Int)] = Array((two,2), (one,1), (three,3))

注意:

  1. 基于当前的实现, groupByKey必须在内存中持有所有的键值对. 如果一个key有太多的value, 则会导致内存溢出(OutOfMemoryError)
  2. 所以这操作非常耗资源, 如果分组的目的是为了在每个key上执行聚合操作(比如: sum 和 average), 则应该使用PairRDDFunctions.aggregateByKey 或者PairRDDFunctions.reduceByKey, 因为他们有更好的性能(会先在分区进行预聚合)

3. reduceByKey(func, [numTasks])

  • 1. 作用

  在一个(K,V)的 RDD 上调用，返回一个(K,V)的 RDD，使用指定的reduce函数，将相同key的value聚合到一起，reduce任务的个数可以通过第二个可选的参数来设置。

  • 2. 案例：创建一个pairRDD，计算相同key对应值的相加结果
代码语言：javascript复制
// 1.创建一个pairRDD
scala> val rdd = sc.parallelize(List(("female",1),("male",5),("female",5),("male",2)))
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[46] at parallelize at <console>:24

// 2.计算相同key对应值的相加结果
scala> val reduce = rdd.reduceByKey((x,y) => x y)
reduce: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[47] at reduceByKey at <console>:26

// 3.打印结果
scala> reduce.collect()
res29: Array[(String, Int)] = Array((female,6), (male,7))

4. reduceByKey和groupByKey的区别

  • reduceByKey：按照key进行聚合，在shuffle之前有combine（预聚合）操作，返回结果是RDD[k,v]。
  • groupByKey：按照key进行分组，直接进行shuffle。
  • 开发指导：reduceByKey比groupByKey性能更好，建议使用。但是需要注意是否会影响业务逻辑。

5. aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])

  • 1. 函数声明：
代码语言：javascript复制
def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U,
                                              combOp: (U, U) => U): RDD[(K, U)] = self.withScope {
    aggregateByKey(zeroValue, defaultPartitioner(self))(seqOp, combOp)
}

  使用给定的 combine 函数和一个初始化的zero value, 对每个key的value进行聚合.   这个函数返回的类型U不同于源 RDD 中的V类型. U的类型是由初始化的zero value来定的. 所以, 我们需要两个操作: - 一个操作(seqOp)去把 1 个v变成 1 个U - 另外一个操作(combOp)来合并 2 个U   第一个操作用于在一个分区进行合并, 第二个操作用在两个分区间进行合并.   为了避免内存分配, 这两个操作函数都允许返回第一个参数, 而不用创建一个新的U

  • 2. 参数描述:
  1. zeroValue：给每一个分区中的每一个key一个初始值；
  2. seqOp：函数用于在每一个分区中用初始值逐步迭代value；
  3. combOp：函数用于合并每个分区中的结果。
  • 3. 案例：创建一个pairRDD，取出每个分区相同key对应值的最大值，然后相加
  • 4. 案例分析：
  • 5. 案例具体过程：计算每个 key 的平均值
代码语言：javascript复制
// 1.创建一个pairRDD
scala> val rdd = sc.parallelize(List(("a",3),("a",2),("c",4),("b",3),("c",6),("c",8)),2)
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[0] at parallelize at <console>:24

// 2.取出每个分区相同key对应值的最大值，分区间然后相加
scala> val agg = rdd.aggregateByKey(0)(math.max(_,_),_ _)
agg: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[1] at aggregateByKey at <console>:26

// 3.打印结果
scala> agg.collect()
res0: Array[(String, Int)] = Array((b,3), (a,3), (c,12))

6. foldByKey

  参数：(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]

  • 1. 作用

  aggregateByKey的简化操作，seqop和combop相同

  • 2. 案例：创建一个pairRDD，计算相同key对应值的相加结果
代码语言：javascript复制
// 1.创建一个pairRDD
scala> val rdd = sc.parallelize(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)),3)
rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[91] at parallelize at <console>:24

// 2.计算相同key对应值的相加结果
scala> val agg = rdd.foldByKey(0)(_ _)
agg: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[92] at foldByKey at <console>:26

// 3.打印结果
scala> agg.collect()
res61: Array[(Int, Int)] = Array((3,14), (1,9), (2,3))

7. combineByKey[C]

  参数：(createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C)

  • 1. 作用

  对相同K，把V合并成一个集合。

  • 2. 参数描述：
  • （1）createCombiner: combineByKey()会遍历分区中的所有元素，因此每个元素的键要么还没有遇到过，要么就和之前的某个元素的键相同。如果这是一个新的元素,combineByKey()会使用一个叫作createCombiner()的函数来创建那个键对应的累加器的初始值
  • （2）mergeValue:如果这是一个在处理当前分区之前已经遇到的键，它会使用mergeValue()方法将该键的累加器对应的当前值与这个新的值进行合并
  • （3）mergeCombiners: 由于每个分区都是独立处理的，因此对于同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器， 就需要使用用户提供的mergeCombiners() 方法将各个分区的结果进行合并。
  • 3. 案例：创建一个pairRDD，根据key计算每种key的均值。（先计算每个key出现的次数以及可以对应值的总和，再相除得到结果）
  • 4. 案例分析：
  • 5. 案例具体过程：对比几个按照 key 聚集的函数的区别和联系
代码语言：javascript复制
// 1.创建一个pairRDD
scala> val input = sc.parallelize(Array(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)),2)
input: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[52] at parallelize at <console>:26

// 2.将相同key对应的值相加，同时记录该key出现的次数，放入一个二元组
scala> val combine = input.combineByKey((_,1),(acc:(Int,Int),v)=>(acc._1 v,acc._2 1),(acc1:(Int,Int),acc2:(Int,Int))=>(acc1._1 acc2._1,acc1._2 acc2._2))
combine: org.apache.spark.rdd.RDD[(String, (Int, Int))] = ShuffledRDD[5] at combineByKey at <console>:28

// 3.打印合并后的结果
scala> combine.collect
res5: Array[(String, (Int, Int))] = Array((b,(286,3)), (a,(274,3)))

// 4.计算平均值
scala> val result = combine.map{case (key,value) => (key,value._1/value._2.toDouble)}
result: org.apache.spark.rdd.RDD[(String, Double)] = MapPartitionsRDD[54] at map at <console>:30

// 5.打印结果
scala> result.collect()
res33: Array[(String, Double)] = Array((b,95.33333333333333), (a,91.33333333333333))

8. sortByKey([ascending], [numTasks])

  • 1. 作用

  在一个(K,V)的 RDD 上调用, K必须实现 Ordered[K] 接口(或者有一个隐式值: Ordering[K]), 返回一个按照key进行排序的(K,V)的 RDD

  • 2. 案例：创建一个pairRDD，按照key的正序和倒序进行排序
代码语言：javascript复制
// 1.创建一个pairRDD
scala> val rdd = sc.parallelize(Array((3,"aa"),(6,"cc"),(2,"bb"),(1,"dd")))
rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[14] at parallelize at <console>:24

// 2.按照key的正序
scala> rdd.sortByKey(true).collect()
res9: Array[(Int, String)] = Array((1,dd), (2,bb), (3,aa), (6,cc))

// 3.按照key的倒序
scala> rdd.sortByKey(false).collect()
res10: Array[(Int, String)] = Array((6,cc), (3,aa), (2,bb), (1,dd))

9. mapValues

  • 1. 作用

  针对于(K,V)形式的类型只对V进行操作

  • 2. 案例：创建一个pairRDD，并将value添加字符串"|||"
代码语言：javascript复制
// 1.创建一个pairRDD
scala> val rdd3 = sc.parallelize(Array((1,"a"),(1,"d"),(2,"b"),(3,"c")))
rdd3: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[67] at parallelize at <console>:24

// 2.对value添加字符串"|||"
scala> rdd3.mapValues(_ "|||").collect()
res26: Array[(Int, String)] = Array((1,a|||), (1,d|||), (2,b|||), (3,c|||))

9. mapValues

  • 1. 作用

  针对于(K,V)形式的类型只对V进行操作

  • 2. 案例：创建一个pairRDD，并将value添加字符串"|||"
代码语言：javascript复制
// 1.创建一个pairRDD
scala> val rdd3 = sc.parallelize(Array((1,"a"),(1,"d"),(2,"b"),(3,"c")))
rdd3: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[67] at parallelize at <console>:24

// 2.对value添加字符串"|||"
scala> rdd3.mapValues(_ "|||").collect()
res26: Array[(Int, String)] = Array((1,a|||), (1,d|||), (2,b|||), (3,c|||))

10. join(otherDataset, [numTasks])

  • 1. 作用

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

  • 2. 案例：创建两个pairRDD，并将key相同的数据聚合到一个元组。
代码语言：javascript复制
// 1.创建第一个pairRDD
scala> val rdd = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[32] at parallelize at <console>:24

// 2.创建第二个pairRDD
scala> val rdd1 = sc.parallelize(Array((1,4),(2,5),(3,6)))
rdd1: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[33] at parallelize at <console>:24

// 3.join操作并打印结果
scala> rdd.join(rdd1).collect()
res13: Array[(Int, (String, Int))] = Array((1,(a,4)), (2,(b,5)), (3,(c,6)))

注意:

  • 1 如果某一个 RDD 有重复的 Key, 则会分别与另外一个 RDD 的相同的 Key进行组合.
  • 2 也支持外连接:leftOuterJoin, rightOuterJoin, and fullOuterJoin.

11. cogroup(otherDataset, [numTasks])

  • 1. 作用

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

  • 2. 案例：创建两个pairRDD，并将key相同的数据聚合到一个迭代器。
代码语言：javascript复制
// 1.创建第一个pairRDD
scala> val rdd = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[37] at parallelize at <console>:24

// 2.创建第二个pairRDD
scala> val rdd1 = sc.parallelize(Array((1,4),(2,5),(3,6)))
rdd1: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[38] at parallelize at <console>:24

// 3.cogroup两个RDD并打印结果
scala> rdd.cogroup(rdd1).collect()
res14: Array[(Int, (Iterable[String], Iterable[Int]))] = Array((1,(CompactBuffer(a),CompactBuffer(4))), (2,(CompactBuffer(b),CompactBuffer(5))), (3,(CompactBuffer(c),CompactBuffer(6))))

  本次的分享就到这里了

spark func key map shuffle

0 人点赞