Transformation转换算子之Value类型

定义SparkContext

  val conf=new SparkConf().setMaster("local[4]").setAppName("test")
  val sc=new SparkContext(conf)

Map算子

  @Test
  def mapTest(): Unit ={
    // 创建本地集合 RDD ,最小分区数为4
    val list=List(1,2,3,4,5,6,7,8)
    val rdd1=sc.makeRDD(list,4)

    // 使用 map 算子
    val mapRdd: RDD[Unit] = rdd1.map(i => {
      println(i)
    })

    // 调用
    mapRdd.collect()
  }

运行结果；以为是分区计算，所以结果是无序的。

5
3
7
1
8
4
6
2

查看每个分区处理哪些数据

rdd1.mapPartitionsWithIndex((index,it)=>{
    println(s"index=$index data=${it.toList}")
    it
}).collect

运行结果

index=3 data=List(7, 8)
index=0 data=List(1, 2)
index=2 data=List(5, 6)
index=1 data=List(3, 4)

获取map算子返回

  @Test
  def mapTest(): Unit ={
    // 创建本地集合 RDD ,最小分区数为4
    val list=List(1,2,3,4,5,6,7,8)
    val rdd1=sc.makeRDD(list,4)

    // 使用 map 算子
    val mapRdd: RDD[Int] = rdd1.map(i =>i*i)

    // 调用
    println(mapRdd.collect().toList)
  }

List(1, 4, 9, 16, 25, 36, 49, 64)

其基本操作和scala的map一致。
map 算子 源码

 def map[U: ClassTag](f: T => U): RDD[U] = withScope {
   // clean 用于清理函数闭包，以为闭包无法进行序列化。
    val cleanF = sc.clean(f)
   // 创建 MapPartitionsRDD 对象；
   // this：创建新的RDD时绑定当前RDD的依赖关系
   // 需要传入一个函数，有三个参数
   // 第一个_：指定TaskContext 
   // 第二个_：指定TaskContext 
   // iter：当前分区的迭代器（内容如下）；然后调用scala中的map而不是spark的map算子。
   //index=3 data=List(7, 8)
   //index=0 data=List(1, 2)
   //index=2 data=List(5, 6)
   //index=1 data=List(3, 4)
    new MapPartitionsRDD[U, T](this, (_, _, iter) => iter.map(cleanF))
  }

思考一个问题？map算子并没有指定分区，但是却是还是4个分区？
首先 map的数据来源于rdd1;rdd1指定了分区。

val rdd1=sc.makeRDD(list,4)

然后map绑定当前rdd的关联关系

// 由rdd1 调用
val mapRdd: RDD[Int] = rdd1.map(i =>i*i)
// 返回一个新的 rdd时绑定了当前的rdd 也就是this
new MapPartitionsRDD[U, T](this, (_, _, iter) => iter.map(cleanF))

最重要的还是 MapPartitionsRDD 继承的RDD就是rdd1 ,可以通过 prev看到。

private[spark] class MapPartitionsRDD[U: ClassTag, T: ClassTag]( 
    var prev: RDD[T],
    f: (TaskContext, Int, Iterator[T]) => Iterator[U], 
    其他参数
） extends RDD[U](prev) {

所以map算子的分区大小是其父类指定的分区大小。

mapPartitions 算子

案例：使用mapPartitions，通过id查询用户信息

  @Test
  def mysqlQueryByMapPartitions(): Unit ={
    // 读取文件
    val lines=sc.textFile("file:///C:/Users/123456/Desktop/userId.txt",4)

    val arr=lines.mapPartitions(it => {
      // 加载驱动类
      Class.forName("com.mysql.jdbc.Driver")

      //连接数据库
      val connection= DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall?characterEncoding=UTF-8", "root", "123321")

      val statement = connection.prepareStatement("select * from user_info where id=?")

      try{
        while (it.hasNext) {
          //主键id
          val id: String = it.next()
          // 设置参数
          statement.setInt(1,id.toInt)
          // 执行查询
          val result: ResultSet = statement.executeQuery()

          while (result.next()) {
            val id = result.getInt("id")
            val nickName = result.getString("nick_name")
            val name = result.getString("name")
            println(s"id=$id,nickName=$nickName,name=$name")
          }
        }

      }catch {
        case e:Exception => {
          println("数据库查询失败")
          println(e.printStackTrace())
        }
      }finally {
        //关闭资源
        if(statement!=null) statement.close()
        if(connection!=null) connection.close()
      }
      it
    }).collect()

    println(arr.toList)

  }

我是将id 一行行存储到该文件中的。数据量不大，一百个id。

val lines=sc.textFile("file:///C:/Users/123456/Desktop/userId.txt",4)

数据就不复制出来了，就看看数据库资源关闭了几次。总共有四个分区，所以最终数据库关闭了4次。

关闭数据库资源....
关闭数据库资源....
关闭数据库资源....
关闭数据库资源....

将mapPartitions换成 map

@Test
  def mysqlQueryByMap(): Unit ={
    // 读取文件
    val lines=sc.textFile("file:///C:/Users/123456/Desktop/userId.txt",4)

    val arr=lines.map(id => {
      // 加载驱动类
      Class.forName("com.mysql.jdbc.Driver")

      //连接数据库
      val connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall?characterEncoding=UTF-8", "root", "123321")

      val statement = connection.prepareStatement("select * from user_info where id=?")

      try {
        // 设置参数
        statement.setInt(1, id.toInt)
        // 执行查询
        val result: ResultSet = statement.executeQuery()

        while (result.next()) {
          val id = result.getInt("id")
          val nickName = result.getString("nick_name")
          val name = result.getString("name")
          println(s"id=$id,nickName=$nickName,name=$name")
        }
      }catch {
        case e:Exception => {
          println("数据库查询失败")
          println(e.printStackTrace())
        }
      }finally {
        println("关闭数据库资源....")
        //关闭资源
        if(statement!=null) statement.close()
        if(connection!=null) connection.close()
      }
    }).collect()
  }

数据库的建立与销毁来来回回一百次(可以自己试试。

关闭数据库资源....
关闭数据库资源....
关闭数据库资源....
关闭数据库资源....
关闭数据库资源....
关闭数据库资源....
关闭数据库资源....
...

数据量少所以，没有问题，但是若数据不是一百而是上百万，千万呢？肯定是不行的。
可能将资源链接丢入map中才会造成这样的原因。
如果把connection 提出去会怎么样？

  @Test
  def mysqlQueryByMap(): Unit ={
    // 读取文件
    val lines=sc.textFile("file:///C:/Users/123456/Desktop/userId.txt",4)

    //连接数据库
    val connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall?characterEncoding=UTF-8", "root", "123321")

    val statement = connection.prepareStatement("select * from user_info where id=?")


    try {

      val arr=lines.map(id => {
        // 加载驱动类
        Class.forName("com.mysql.jdbc.Driver")
        // 设置参数
        statement.setInt(1, id.toInt)
        // 执行查询
        val result: ResultSet = statement.executeQuery()

        while (result.next()) {
          val id = result.getInt("id")
          val nickName = result.getString("nick_name")
          val name = result.getString("name")
          println(s"id=$id,nickName=$nickName,name=$name")
        }

      }).collect()

    }catch {
      case e:Exception => {
        println("数据库查询失败")
        println(e.printStackTrace())
      }
    }finally {
      println("关闭数据库资源....")
      //关闭资源
      if(statement!=null) statement.close()
      if(connection!=null) connection.close()
    }
  }

报了一个错，该错误的原因是jdbc，并没有实现序列化，无法进行传输。

org.apache.spark.SparkException: Task not serializable

该案例除了说明 map 与 mapPartitions 的区别外，更想表达的意思是。
rdd 会将数据进行分区，每个分区的计算逻辑或数据可能不在同一个节点上。即使是local模式，分区之间也是并行处理。

mapPartitions 与 map 的区别：

  1. map里面的函数是针对分区里面的每个元素进行计算，mapPartitions里面的函数是针对每个分区的所有数据的迭代器进行计算
  2. map里面的函数是计算一个元素返回一个结果,所以map生成的新的RDD里面的元素个数 = 原来RDD元素个数
    mapPartitions里面的函数是计算一个分区的所有数据的迭代器然后返回一个新的迭代器,所以mapPartitions生成的新的RDD里面的元素的个数与原来RDD元素个数可能不相同
  3. map是针对单个元素操作,元素操作完成之后可以进行回收内存了
    mapPartitions是针对一个迭代器操作，操作完成迭代器一个元素之后,该元素不能回收必须等到整个迭代器都处理完成之后才能回收。如果一个分区中数据量很大,可能导致内存溢出。如果出现内存溢出可以用map代替。【完成比完美更重要】

mapPartitions源码赏析

  // f：传入一个函数，参为迭代器
  // preservesPartitioning:是否保留分区，默认为false
  def mapPartitions[U: ClassTag](f: Iterator[T] => Iterator[U],preservesPartitioning: Boolean = false): RDD[U] = withScope {
    // 清除闭包，保证数据可以进行序列化传输
    val cleanedF = sc.clean(f)
    // 创建 MapPartitionsRDD
    // this 绑定当前RDD,
    // iter 迭代器
    new MapPartitionsRDD(this,(_: TaskContext, _: Int, iter: Iterator[T]) => cleanedF(iter),preservesPartitioning)
  }

点击进入MapPartitionsRDD 它会执行一个compute函数

  override def compute(split: Partition, context: TaskContext): Iterator[U] =
    f(context, split.index, firstParent[T].iterator(split, context))

mapPartitionsWithIndex

和 mapPartitions类似，但是它可以指定分区号

  @Test
  def mapPartitionsWithIndexTest(): Unit ={
    val list=List(1,2,3,4,5,6,7,8)
    val rdd1=sc.makeRDD(list,4)
    rdd1.mapPartitionsWithIndex((index,it)=>{
      while (it.hasNext) {
        println(s"index=$index, ${it.next()}")
      }
      it
    }).collect()
  }

结果

index=2, 5
index=1, 3
index=0, 1
index=1, 4
index=2, 6
index=0, 2
index=3, 7
index=3, 8

flatMap()

与map操作类似，将RDD中的每一个元素通过应用f函数依次转换为新的元素，并封装到RDD中。
区别：在flatMap操作中，f函数的返回值是一个集合，并且会将每一个该集合中的元素拆分出来放到新的RDD中。

  @Test
  def flatMapTest(): Unit ={
    val list=List(
      "hello,java,scala",
      "python,html,xml",
      "xpath,js,vue",
      "linux,windows"
    )
    // 创建本地集合RDD
    val rdd1= sc.parallelize(list, 4)
    // flatMap
    val rdd2= rdd1.flatMap(_.split(","))
    // 计算，汇总
    println(rdd2.collect.toList)
  }

结果

List(hello, java, scala, python, html, xml, xpath, js, vue, linux, windows)

glom()

该操作将RDD中每一个分区变成一个数组，并放置在新的RDD中，数组中元素的类型与原分区中元素类型一致

  @Test
  def glomTest(): Unit ={

    val list: Seq[Int] =List(1,2,3,4,5,6,7,8)
    val rdd1: RDD[Int] =sc.parallelize(list,4)
    val rdd2: RDD[Array[Int]] =rdd1.glom()
    // 获取
    val arrList: List[Array[Int]] = rdd2.collect.toList
    for (arr<- arrList){
      println(arr.toList)
    }

  }

List(1, 2)
List(3, 4)
List(5, 6)
List(7, 8)

groupBy

对数据进行分组

  @Test
  def groupBy(): Unit ={
    // 生成一百个数
    val range = Range(0, 100)

    val rdd1: RDD[Int] = sc.parallelize(range, 4)


    // 将一百以内的数据按照 2的倍数和3的倍数 进行分类。
    val f=(i:Int)=>{
      if(i%2==0 && i%3==0){0}
      else if(i%2==0){1}
      else if(i%3==0){2}
      else -1
    }


    val rdd2: RDD[(Int, Iterable[Int])] = rdd1.groupBy(f)

    val result: List[(Int, Iterable[Int])] = rdd2.collect.toList
    for(r <- result){
        r match {
          case (k,v) => println(k,v)
        }
    }

  }

结果

(0,CompactBuffer(0, 6, 12, 18, 24, 30, 36, 42, 48, 54, 60, 66, 72, 78, 84, 90, 96))
(1,CompactBuffer(2, 4, 8, 10, 14, 16, 20, 22, 26, 28, 32, 34, 38, 40, 44, 46, 50, 52, 56, 58, 62, 64, 68, 70, 74, 76, 80, 82, 86, 88, 92, 94, 98))
(2,CompactBuffer(3, 9, 15, 21, 27, 33, 39, 45, 51, 57, 63, 69, 75, 81, 87, 93, 99))
(-1,CompactBuffer(1, 5, 7, 11, 13, 17, 19, 23, 25, 29, 31, 35, 37, 41, 43, 47, 49, 53, 55, 59, 61, 65, 67, 71, 73, 77, 79, 83, 85, 89, 91, 95, 97))

使用groupby 完成worldCount作业

@Test
  def worldCount():Unit={
    //读取文件
    val lines=sc.textFile("file:///C:/Users/123456/Desktop/worldCount.txt",4)

    // 内容扁平化
    val worldList: RDD[String] = lines.flatMap(_.split(" "))

    // 内容分组
    val groupList: RDD[(String, Iterable[String])] = worldList.groupBy(s => s)

    // 统计单词数量
    val result=groupList.map(x=>(x._1,x._2.size))

    println(result.collect().toList)

  }

数据结果

List((shell,4), (wahaha,1), (hello,2), (python,1), (java,5))

filter 过滤

接收一个返回值为布尔类型的函数作为参数。当某个RDD调用filter方法时，会对该RDD中每一个元素应用f函数，如果返回值类型为true，则该元素会被添加到新的RDD中。

案例：找到一百中所有的偶数

  @Test
  def filterTest(): Unit ={
    // 生成0-100的数
    val range=Range(0,100)
    val rdd1: RDD[Int] = sc.parallelize(range, 4)

    //过滤奇数
    val value: RDD[Int] = rdd1.filter(_%2==0)
    println(value.collect.toList)
  }

List(0, 2, 4, 6, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32, 34, 36, 38, 40, 42, 44, 46, 48, 50, 52, 54, 56, 58, 60, 62, 64, 66, 68, 70, 72, 74, 76, 78, 80, 82, 84, 86, 88, 90, 92, 94, 96, 98)

sample 采样

  def sample(withReplacement: Boolean,fraction: Double,seed: Long = Utils.random.nextLong): RDD[T] = {...}
  • withReplacement:
    是否放回[true-代表放回,意味着同一个数据可能被多次采样 false-不返回,意味着同一条数据最多被采样一次] 【工作中设置为false】
  • fraction: 【工作中一般设置为 0.1-0.2 】:
    如果withReplacement=false, fraction代表每个元素被采样的概率[0,1]
    如果withReplacement=true, fraction代表每个元素期望被采样的次数
  • seed: 随机数种子

对应 0-100的数字进行采样；100%

  @Test
  def sampleTest(): Unit ={
    // 生成0-100的数
    val range=Range(0,100)
    val rdd1: RDD[Int] = sc.parallelize(range, 4)
    // withReplacement=false ;fraction=1
    val value: RDD[Int] = rdd1.sample(false, 1)

    val list=value.collect.toList
    println(list)
  }

List(0, 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, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99)

对应 0-100的数字进行采样；50%

  @Test
  def sampleTest(): Unit ={
    // 生成0-100的数
    val range=Range(0,100)
    val rdd1: RDD[Int] = sc.parallelize(range, 4)
    // withReplacement=false ;fraction=0.5
    val value: RDD[Int] = rdd1.sample(false, 0.5)

    val list=value.collect.toList
    println(list)
  }

随机抽查50个不重样数据

List(0, 1, 3, 4, 6, 7, 9, 11, 15, 16, 17, 20, 22, 23, 25, 27, 30, 32, 33, 34, 35, 36, 37, 38, 40, 42, 44, 45, 46, 48, 50, 54, 56, 60, 63, 65, 66, 67, 70, 71, 73, 75, 77, 79, 80, 81, 82, 83, 85, 89, 91, 92, 97)

对应 0-100的数字进行采样；10%

  @Test
  def sampleTest(): Unit ={
    // 生成0-100的数
    val range=Range(0,100)
    val rdd1: RDD[Int] = sc.parallelize(range, 4)
    // withReplacement=false ;fraction=0.1
    val value: RDD[Int] = rdd1.sample(false, 0.1)

    val list=value.collect.toList
    println(list)
  }

这样看起方便点

List(15, 23, 44, 49, 50, 83, 85, 86, 96)

重复采样，单个元素可能被采样多次

  @Test
  def sampleTest(): Unit ={
    // 生成0-100的数
    val range=Range(0,100)
    val rdd1: RDD[Int] = sc.parallelize(range, 4)
    // withReplacement=true ;fraction=0.2 
    val value: RDD[Int] = rdd1.sample(true, 0.2)

    val list=value.collect.toList
    println(list)
  }

withReplacement 设置为true，有些数据可能会被多次采样（如88，95）。

List(3, 19, 25, 26, 27, 30, 30, 33, 38, 55, 60, 62, 63, 66, 72, 77, 88, 88, 93, 94, 95, 95)

将withReplacement 依旧为true,fraction改为整数
这次采用0-10的数据

  @Test
  def sampleTest(): Unit ={
    // 生成0-100的数
    val range=Range(0,10)
    val rdd1: RDD[Int] = sc.parallelize(range, 4)
    // withReplacement=true ;fraction=2
    val value: RDD[Int] = rdd1.sample(true,2)

    val list=value.collect.toList
    println(list)
  }

如果withReplacement=true, fraction代表每个元素期望被采样的次数；
这个期望值是不确定的，如上我期望值是2但是有的却只有1个如（7）；但也比2高的。

List(0, 0, 0, 0, 0, 1, 1, 1, 3, 3, 4, 4, 5, 5, 5, 5, 5, 5, 7, 8, 8, 8, 9, 9, 9)

最后再说说 seed；默认是一个伪随机数，用来决定采样的随机率。
一般不会更改

Long = Utils.random.nextLong

若seed 固定多次采用的结果也是一样

  @Test
  def sampleTest(): Unit ={
    // 生成0-100的数
    val range=Range(0,10)
    val rdd1: RDD[Int] = sc.parallelize(range, 4)
    // withReplacement=true ;fraction=2;seed=10
    val value: RDD[Int] = rdd1.sample(true,2,10)

    val list=value.collect.toList
    println(list)
  }

第一次运行

List(1, 2, 2, 2, 4, 4, 4, 4, 4, 5, 5, 5, 5, 6, 6, 7, 7, 7, 8, 9, 9, 9, 9)

第二次运行

List(1, 2, 2, 2, 4, 4, 4, 4, 4, 5, 5, 5, 5, 6, 6, 7, 7, 7, 8, 9, 9, 9, 9)

distinct 去重

  @Test
  def distinctTest(): Unit ={
    // 设置一些重复的元素
    val list=List(1,2,2,3,3,4,3,5,6,7,9,8,9,4,7)

    // 创建本地集合RDD
    val rdd1: RDD[Int] = sc.parallelize(list, 4)

    // 去重
    val value: RDD[Int] = rdd1.distinct()

    // 调用，打印
    println(value.collect.toList)
  }

结果

List(4, 8, 1, 9, 5, 6, 2, 3, 7)

除了使用distinct也可使用groupBy 实现去重功能

 @Test
  def distinctTest(): Unit ={
    // 设置一些重复的元素
    val list=List(1,2,2,3,3,4,3,5,6,7,9,8,9,4,7)

    // 创建本地集合RDD
    val rdd1: RDD[Int] = sc.parallelize(list, 4)

    // 去重
    val value: RDD[Int] = rdd1.groupBy(x=>x).map(_._1)

    // 调用，打印
    println(value.collect.toList)
  }

结果

List(4, 8, 1, 9, 5, 6, 2, 3, 7)

coalesce 合并分区

  @Test
  def coalesceTest(): Unit ={
    // 生成0-100的数
    val range=Range(0,100)

    // 创建本地集合RDD
    val rdd1: RDD[Int] = sc.parallelize(range, 4)    
  }

查看分区情况

rdd1.mapPartitionsWithIndex((index,it)=>{
  println(s"index=$index;it=${it.toList}")
  it
}).collect

各个分区情况

index=0;it=List(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24)
index=1;it=List(25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49)
index=3;it=List(75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99)
index=2;it=List(50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74)

合并分区
numPartitions：合并分区的个数
shuffle：默认为false，并不会触发shuffle，设置为true，可以重新扩大分区，但是会进行shuffle操作。

 def coalesce(numPartitions: Int, shuffle: Boolean = false,
               partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
              (implicit ord: Ordering[T] = null): RDD[T] = withScope {...}

合并成两个分区，并查看合并分区后的数据情况

    //合并成两个分区
    val value: RDD[Int] = rdd1.coalesce(2)
    println(s"分区数${value.getNumPartitions}")
    //查看合并之后的分区情况
    value.mapPartitionsWithIndex((index,it)=>{
      println(s"index=$index;it=${it.toList}")
      it
    }).collect

结果

分区数2
index=0;it=List(0, 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, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49)
index=1;it=List(50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99)

默认情况下，无法增加分区

    //合并成两个分区
    val value: RDD[Int] = rdd1.coalesce(5)
    println(s"分区数${value.getNumPartitions}")
    //查看合并之后的分区情况
    value.mapPartitionsWithIndex((index,it)=>{
      println(s"index=$index;it=${it.toList}")
      it
    }).collect

就打印分区数就行了，无法扩展分区。

分区数2

coalesce默认情况下只能合并分区,如果想要增大分区数,需要设置shuffle=true

    //合并成两个分区
    val value: RDD[Int] = rdd1.coalesce(5,true)
    println(s"分区数${value.getNumPartitions}")

    //查看合并之后的分区情况
    value.mapPartitionsWithIndex((index,it)=>{
      println(s"index=$index;it=${it.toList}")
      it
    }).collect

数据结果

分区数5
index=0;it=List(4, 9, 14, 19, 24, 27, 32, 37, 42, 47, 53, 58, 63, 68, 73, 78, 83, 88, 93, 98)
index=1;it=List(0, 5, 10, 15, 20, 28, 33, 38, 43, 48, 54, 59, 64, 69, 74, 79, 84, 89, 94, 99)
index=3;it=List(2, 7, 12, 17, 22, 25, 30, 35, 40, 45, 51, 56, 61, 66, 71, 76, 81, 86, 91, 96)
index=2;it=List(1, 6, 11, 16, 21, 29, 34, 39, 44, 49, 50, 55, 60, 65, 70, 75, 80, 85, 90, 95)
index=4;it=List(3, 8, 13, 18, 23, 26, 31, 36, 41, 46, 52, 57, 62, 67, 72, 77, 82, 87, 92, 97)

源码解析

def coalesce(numPartitions: Int, shuffle: Boolean = false,partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
          (implicit ord: Ordering[T] = null): RDD[T] = withScope {
    require(numPartitions > 0, s"Number of partitions ($numPartitions) must be positive.")
    // 判断 shuffle 是否为true
    if (shuffle) {
      /** Distributes elements evenly across output partitions, starting from a random partition. */
      val distributePartition = (index: Int, items: Iterator[T]) => {
        var position = new Random(hashing.byteswap32(index)).nextInt(numPartitions)
        items.map { t =>
          // Note that the hash code of the key will just be the key itself. The HashPartitioner
          // will mod it with the number of total partitions.
          position = position + 1
          (position, t)
        }
      } : Iterator[(Int, T)]

      // include a shuffle step so that our upstream tasks are still distributed
      new CoalescedRDD(
        new ShuffledRDD[Int, T, T](
          mapPartitionsWithIndexInternal(distributePartition, isOrderSensitive = true),
          new HashPartitioner(numPartitions)),
        numPartitions,
        partitionCoalescer).values
    } else {
      new CoalescedRDD(this, numPartitions, partitionCoalescer)
    }
}

先分析shuffle=false情况

// this 当前调用coalesce的rdd
// numPartitions 分区数 ；上面设置的是5
new CoalescedRDD(this, numPartitions, partitionCoalescer)

partitionCoalescer 我们并没有指定，所以使用的是默认的DefaultPartitionCoalescer

  override def getPartitions: Array[Partition] = {
    val pc = partitionCoalescer.getOrElse(new DefaultPartitionCoalescer())

    pc.coalesce(maxPartitions, prev).zipWithIndex.map {
      case (pg, i) =>
        val ids = pg.partitions.map(_.index).toArray
        CoalescedRDDPartition(i, prev, ids, pg.prefLoc)
    }
  }
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 218,640评论 6赞 507
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 93,254评论 3赞 395
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 165,011评论 0赞 355
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 58,755评论 1赞 294
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 67,774评论 6赞 392
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 51,610评论 1赞 305
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 40,352评论 3赞 418
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 39,257评论 0赞 276
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 45,717评论 1赞 315
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,894评论 3赞 336
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 40,021评论 1赞 350
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 35,735评论 5赞 346
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 41,354评论 3赞 330
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,936评论 0赞 22
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 33,054评论 1赞 270
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 48,224评论 3赞 371
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 44,974评论 2赞 355

推荐阅读更多精彩内容