数值类型转换算子

1.map算子

• 定义: map(fun)函数
• 解释: 将RDD的每一个元素通过fun函数计算得到一个新的结果，新的结果我们会组合成为一个新的RDD
• 特别注意：一对一场景下，RDD的每一条数据对应新的RDD的中一条数据
• 案例：

def mapOper(sc: SparkContext): Unit = {
  println("----------------map1开始------------------")
  val rdd = sc.makeRDD(Array(1, 2, 3, 4, 5, 0), 1)
  val map = rdd.map(_ * 3)
  map.foreach(println(_)) // 结果： 3  6  9  12  15  0
  println("----------------map1结束------------------")

  println("----------------map2开始------------------")
  val rdd2 = sc.makeRDD(Array("hbase", "flink", "kafka"))
  val word = rdd2.map((_, 1))
  word.foreach(println(_)) // 结果： (hbase,1)  (flink,1)  (kafka,1)
  println("----------------map2结束------------------")
}

2.filter过滤算子

• 定义: filter(fun)算子
• 解释：将RDD中每一个元素通过fun函数计算得到一个boolean类型的返回值，如果返回值为true保留数据
• 特别注意：如果返回值为false，那么数据就舍弃 得到的新的RDD的类型和旧的RDD类型一样的
• 案例：

def filterOper(sc: SparkContext): Unit = {
  println("----------------filter开始------------------")
  /**
   * 1 to 100 是scala中Seq集合的一种创建语法
   */
  val rdd = sc.makeRDD(1 to 10)
  val filter = rdd.filter((num: Int) => {
    if (num % 2 == 0) {
      true
    } else {
      false
    }
  })
  filter.foreach(println(_)) // 结果： 2  6  4  8  10
  println("----------------filter结束------------------")
}

3.flatMap压扁算子

• 定义：flatMap(fun)算子
• 解释：原先的RDD一条数据经过flatMap算子操作返回一个集合数据，集合数据中每一条数据就是新的RDD中的数据
• 特别注意：一对多
• 案例：

def flatMap(sc: SparkContext): Unit = {
  println("----------------flatMap开始------------------")
  val rdd = sc.makeRDD(Array("i am people", "dog is an animal"))
  val flatmap = rdd.flatMap((str: String) => {
    val word: Array[String] = str.split(" ")
    word
  })
  flatmap.foreach(println(_)) // 结果： i  am  people  dog  is  an  animal
  println("----------------flatMap结束------------------")
}

4.mapPartitions

• 定义：mapPartitions(fun)算子
• 解释：和map算子函数含义一致，都是一个分区数据经过fun函数得到一个新的分区数据结果，新的结果组成新的RDD
• 案例：

def mapPartitionsOper(sc: SparkContext): Unit = {
  println("----------------mapPartitions开始------------------")
  val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5, 6, 7, 8), 5)
  val mapPartitions = rdd.mapPartitions((list: Iterator[Int]) => {
    val listBuffer: ListBuffer[Int] = ListBuffer()
    for (num <- list) {
      println(s"处理的分区数据是 $num")
      listBuffer.append(num * 3)
    }
    listBuffer.iterator
  })
  /*
    处理的分区数据是 1
    处理的分区数据是 2
    3
    处理的分区数据是 3
    6
    9
    处理的分区数据是 4
    12
    处理的分区数据是 5
    处理的分区数据是 6
    15
    18
    处理的分区数据是 7
    处理的分区数据是 8
    21
    24
  */
  mapPartitions.foreach(println(_))
  println("----------------mapPartitions结束------------------")
}

5.mapPartitionsWithIndex

• 定义：mapPartitionsWithIndex(fun)算子
• 解释：一次处理一个分区的数据，最后返回一个迭代器，附带传入一个分区索引
  fun函数输入参数类型是一个2元组（Int,Iterator[T]）=>Iterator[U] :RDD[U]
• 案例：

def mapPartitionsWithIndexOper(sc: SparkContext): Unit = {
  println("----------------mapPartitionsWithIndex开始------------------")
  val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5, 6, 7, 8), 5)
  val mapPartitionsWithIndex: RDD[Int] = rdd.mapPartitionsWithIndex((index: Int, iterator: Iterator[Int]) => {
    val listBuffer: ListBuffer[Int] = ListBuffer()
    for (num <- iterator) {
      println(s"分区为 $index , 分区的数据为 $num ")
      listBuffer.append(num)
    }
    listBuffer.iterator
  })
  /*
    分区为 1 , 分区的数据为 2 
    分区为 0 , 分区的数据为 1 
    分区为 1 , 分区的数据为 3 
    1
    2
    3
    分区为 2 , 分区的数据为 4 
    4
    分区为 3 , 分区的数据为 5 
    分区为 3 , 分区的数据为 6 
    5
    6
    分区为 4 , 分区的数据为 7 
    分区为 4 , 分区的数据为 8 
    7
    8
  */
  mapPartitionsWithIndex.foreach(println((_)))
  println("----------------mapPartitionsWithIndex结束------------------")
}

总结

如果要实现将一个RDD数据集中每一条数据转换成为一个新的RDD数据集中记录，
可以使用map、mapPartitions、mapPartitiosWithIndex三种方式实现
一般我们建议使用mapPartitions、mapPartitiosWithIndex 这样的效率比较高点

6.sample

• 定义：sample(withReplacement, fraction, seed):RDD[T] 抽样算子
• 解释：抽样一部分数据进行处理分析（每个分区都会抽取数据）
• 案例：