https://buwenbuhuo.blog.csdn.net/category_10163099_2.html

在 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类型(其实就是存一个二维的元组)

详细操作见 Idea上

1 map(func)

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

image.png

  1. 案例：创建一个包含1-10的的 RDD，然后将每个元素*2形成新的 RDD ```scala scala > val rdd1 = sc.parallelize(1 to 10) rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at :24

// 得到一个新的 RDD, 但是这个 RDD 中的元素并不是立即计算出来的 scala> val rdd2 = rdd1.map(_ * 2) rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[1] at map at

:26

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

  2. <a name="MaBpz"></a>
  3. ## 2 mapPartitions(func)
  4. 1. 作用<br />类似于map(func), 但是是独立在每个分区上运行.所以：Iterator<T> => Iterator<U><br />假设有N个元素，有M个分区，那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区。
  6. 2. 案例：创建一个RDD，使每个元素*2组成新的RDD
  7. ```scala
  8. // 1. 创建一个RDD
  9. scala> val source = sc.parallelize(1 to 10)
  10. source: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at parallelize at <console>:24
  12. // 2. 使每个元素*2组成新的RDD
  13. scala> source.mapPartitions(it => it.map(_ * 2))
  14. res7: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at mapPartitions at <console>:27
  16. // 3. 打印新的RDD
  17. scala> res7.collect
  18. 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) => Iterator
    2. 案例：创建一个RDD，使每个元素跟所在分区形成一个元组组成一个新的RDD ```scala // 1.创建一个RDD scala> val rdd = sc.parallelize(Array(1,2,3,4)) rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at :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 :26

// 3.打印新的RDD scala> indexRdd.collect res2: Array[(Int, Int)] = Array((0,1), (0,2), (1,3), (1,4))

  1. 3. 分区数的确定, 和对数组中的元素如何进行分区
  2. ```scala
  3. // 1.确定分区数:
  4. override def defaultParallelism(): Int = scheduler.conf.getInt("spark.default.parallelism", totalCores)
  6. // 2.对元素进行分区
  7. // length: RDD 中数据的长度  numSlices: 分区数
  8. def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {
  9.  (0 until numSlices).iterator.map { i =>val start = ((i * length) / numSlices).toInt
  10.    val end = (((i + 1) * length) / numSlices).toInt
  11.    (start, end)
  12.  }
  13. }
  14. seq match {
  15.  case r: Range =>
  17.  case nr: NumericRange[_] =>
  19.  case _ =>
  20.    val array = seq.toArray // To prevent O(n^2) operations for List etc
  21.    positions(array.length, numSlices).map { case (start, end) =>
  22.        array.slice(start, end).toSeq
  23.    }.toSeq
  24. }

4 map()和mapPartitions()的区别

  • map()：每次处理一条数据。
  • mapPartitions()：每次处理一个分区的数据，这个分区的数据处理完后，原 RDD 中该分区的数据才能释放，可能导致 OOM。

  • 开发指导：当内存空间较大的时候建议使用mapPartitions()，以提高处理效率。

    5 flatMap(func)：

    flatMap的执行结果跟参数有关，（1->1,2->1,2……5->1,2,3,4,5）有三个参数那么每个元素都会被映射成三个元素

    1. 作用：类似于map，但是每一个输入元素可以被映射为 0 或多个输出元素（所以func应该返回一个序列，而不是单一元素 T => TraversableOnce[U]）
      2. 案例1：创建一个元素为1-5的RDD，运用flatMap创建一个新的RDD，新的RDD为原RDD的每个元素的2倍（2，4，6，8，10） ```scala // 1.创建 scala> val sourceFlat = sc.parallelize(1 to 5) sourceFlat: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[12] at parallelize at :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 :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)

  1.  3. 案例2：创建一个元素为 1-5 的RDD，运用 flatMap创建一个新的 RDD，新的 RDD 为原 RDD 每个元素的 平方和三次方 来组成 1,1,4,8,9,27…
  2. ```scala
  4. scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5))
  5. rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at parallelize at <console>:24
  7. scala> rdd1.flatMap(x => Array(x * x, x * x * x))
  8. res13: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at flatMap at <console>:27
  10. scala> res13.collect
  11. res14: Array[Int] = Array(1, 1, 4, 8, 9, 27, 16, 64, 25, 125)

6 glom()

  1. 作用 将每一个分区的元素合并成一个数组，形成新的 RDD 类型是RDD[Array[T]]
    2. 案例：创建一个 4 个分区的 RDD，并将每个分区的数据放到一个数组 ```scala // 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 :24

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

  1. <a name="wWrnK"></a>
  2. ## 7 groupBy(func)
  3.  1. 作用    按照func的返回值进行分组.<br />func返回值作为 key, 对应的值放入一个迭代器中. 返回的 RDD: RDD[(K, Iterable[T])，每组内元素的顺序不能保证, 并且甚至每次调用得到的顺序也有可能不同.
  5.  2. 案例1：创建一个RDD，按照元素模以2的值进行分组。
  6. ```scala
  7. // 1.创建
  8. scala> val rdd = sc.parallelize(1 to 4)
  9. rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[65] at parallelize at <console>:24
  10. // 2.按照元素模以2的值进行分组
  11. scala> val group = rdd.groupBy(_%2)
  12. group: org.apache.spark.rdd.RDD[(Int, Iterable[Int])] = ShuffledRDD[2] at groupBy at <console>:26
  13. // 3.打印结果
  14. scala> group.collect
  15. res0: Array[(Int, Iterable[Int])] = Array((0,CompactBuffer(2, 4)), (1,CompactBuffer(1, 3)))

[

](https://blog.csdn.net/qq_16146103/article/details/107461368)
[

](https://blog.csdn.net/qq_16146103/article/details/107461368)