Spark-RDD 转换算子（Value 类型）

Posted 2023-03-11 open_test01

        

1、map

2、mapPartitions 

3、mapPartitionsWithIndex

4、flatMap

5、glom

6、groupBy 

7、filter

8、sample

9、distinct

10、coalesce

11、repartition

12、sortBy

---

        转换算子其实就是RDD中对数据进行封装转换的方法。

        RDD 根据数据处理方式的不同将算子整体上分为 Value 类型、双 Value 类型和 Key-Value 类型。

1、map

        将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。

def map[U: ClassTag](f: T => U): RDD[U]

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5,6))
    //算子map
      //转换函数
    def mapFun(num:Int): Int = //完成乘2操作
      num * 2
    

    //map（传入的参数为函数）
    val mpRDD: RDD[Int] = rdd.map(mapFun)

    mpRDD.collect().foreach(println) //输出检验
    //关闭环境
    sc.stop()
  

        但是一般情况方法内参数函数使用致简原则匿名函数作为参数

val mpRDD: RDD[Int] = rdd.map(_*2) //完成乘2操作

        map体现RDD的并行计算 

 def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5,6),1)//一个分区

    //1、rdd计算一个分区内的数据一个一个执行逻辑
    //当一个数据的流程全部执行后才会执行下一条数据 分区内数据的执行是有序的

    //2、不同分区的数据分区的计算是无序的
    val mpRDD1: RDD[Int] = rdd.map(
      num => 
        println("num操作1=>"+num)
        num
      
    )

val mpRDD2: RDD[Int] = mpRDD1.map(
num => 
  println("num操作2=>"+num)
  num

  )

  mpRDD2.collect()
  //关闭环境
  sc.stop()

         当分区为1时 

        当分区为2时 

2、mapPartitions 

        将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处 理，哪怕是过滤数据。 

def mapPartitions[U: ClassTag](
 f: Iterator[T] => Iterator[U],
 preservesPartitioning: Boolean = false): RDD[U]

• 可以以分区为单位作为数据转换操作
• 缺点：但是它会把分区的数据加载到内存中进行引用 如果处理完了数据它不会释放内存因为它存在对象的引用。
•  内存较小 数据量较大容易出现内存溢出的情况

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5,6),2)//2个分区

    //mapPartitions(迭代器) 一次性处理一个分区的数据处理完成后再处理下一个分区的数据
    val mapRDD: RDD[Int] = rdd.mapPartitions(
      iter => 
        println("------") //输出现几次则有几个分区
        iter.map(_ * 2)
      
    )
    mapRDD.collect().foreach(println)
  //关闭环境
  sc.stop()

3、mapPartitionsWithIndex

        将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处 理，哪怕是过滤数据，在处理时同时可以获取当前分区索引。

def mapPartitionsWithIndex[U: ClassTag](
 f: (Int, Iterator[T]) => Iterator[U],
 preservesPartitioning: Boolean = false): RDD[U]

        只保留打印第二个分区的数据 

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5,6),2)//2个分区

    //需求功能： 只保留打印第二个分区的数据
    val mpiRDD: RDD[Int] = rdd.mapPartitionsWithIndex(
      (index, iter) => 
        if (index == 1)  //打印索引为1的分区 即第二个分区
          iter
         else 
          Nil.iterator
        
      
    )
    mpiRDD.collect().foreach(println)
  //关闭环境
  sc.stop()

         显示改数据属于哪个分区

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5,6),4)//4个分区

    //需求功能：显示改数据属于哪个分区
    val mpiRDD = rdd.mapPartitionsWithIndex(
      (index, iter) => 
        iter.map(
          num => 
            ("分区号"+index+"数据："+num)
          
        )
      
    )
    mpiRDD.collect().foreach(println)
  //关闭环境
  sc.stop()
  

 4、flatMap

         将处理的数据进行扁平化后再进行映射处理，所以算子也称之为扁平映射

def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    val rdd: RDD[List[Int]] = sc.makeRDD(List(List(1,2),List(3,4)))
    //把rdd中的List中的List分成个体
    val fltRDD: RDD[Int] = rdd.flatMap(
      list => 
        list
      
    )
    fltRDD.collect().foreach(println)
  //关闭环境
  sc.stop()
  

5、glom

        将同一个分区的数据直接转换为相同类型的内存数组进行处理，分区不变 

def glom(): RDD[Array[T]]

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2) //分区：2
    //List => Int
    //Int => Array
    val gRDD: RDD[Array[Int]] = rdd.glom()

    gRDD.collect().foreach(i => println(i.mkString(",")))
  //关闭环境
  sc.stop()
  

        计算所有分区最大值求和（分区内取最大值，分区间最大值求和） 

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2) //分区：2

    val grdd: RDD[Array[Int]] = rdd.glom()
    val maxRDD: RDD[Int] = grdd.map(
      arr => 
        arr.max
      
    )
    println(maxRDD.collect().sum)

  //关闭环境
  sc.stop()
  

6、groupBy 

        将数据根据指定的规则进行分组, 分区默认不变，但是数据会被打乱重新组合，我们将这样 的操作称之为 shuffle。极限情况下，数据可能被分在同一个分区中

        一个组的数据在一个分区中，但是并不是说一个分区中只有一个组

def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2) //分区：2
    //groupBy会将数据源中每一个数据进行分组判断，根据返回的分组Key进行分组
    //相同的Key数据会放置在一个组中
      //groupBy(参数为分组条件的函数)
    val gpRDD: RDD[(Int, Iterable[Int])] = rdd.groupBy(
      num => num % 2
    )
    gpRDD.collect().foreach(println)

  //关闭环境
  sc.stop()
  

 7、filter

        将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。 当数据进行筛选过滤后，分区不变

        但是分区内的数据可能不均衡，生产环境下，可能会出 现数据倾斜。 

def filter(f: T => Boolean): RDD[T]

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

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

    //需求：过滤偶数留下奇数
    val fRDD: RDD[Int] = rdd.filter(num => num%2 !=0)

    fRDD.collect().foreach(println)

  //关闭环境
  sc.stop()
  

8、sample

        根据指定的规则从数据集中抽取数据

def sample(
 withReplacement: Boolean, //抽取完受否放回
 fraction: Double, //比例
 seed: Long = Utils.random.nextLong): RDD[T] //随机数种子（基准值）
 )

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5,6,7,8,9,10))

    //sample(抽取后是否放回，每条数据被抽取的概率,随机算法的种子)
    //如果传染第3个参数则每次抽取的数据是固定的 不传的话就是使用当前系统时间则是随机的
    println(rdd.sample(
      false,
      0.5,
      1
    ).collect().mkString(","))
    
  //关闭环境
  sc.stop()
  

9、distinct

        将数据集中重复的数据去重

def distinct()(implicit ord: Ordering[T] = null): RDD[T]
def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5,1,2,3,4,5))
    //去重
    val dRDD = rdd.distinct()
    dRDD.collect().foreach(println)

  //关闭环境
  sc.stop()
  

10、coalesce

        根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率 当 spark 程序中，存在过多的小任务的时候，可以通过 coalesce 方法，收缩合并分区，减少 分区的个数，减小任务调度成本 

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

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),4) //4个分区
    //缩减分区可能会导致数据倾斜
    //需要使数据均衡则要使用shuffle处理
    //coalesce(参数为分区数量,是否进行shuffle处理)
    val newRDD: RDD[Int] = rdd.coalesce(2,true) //合并为2个分区

    newRDD.saveAsTextFile("output")
  //关闭环境
  sc.stop()
  

11、repartition

        该操作内部其实执行的是 coalesce 操作，参数 shuffle 的默认值为 true。无论是将分区数多的 RDD 转换为分区数少的 RDD，还是将分区数少的 RDD 转换为分区数多的 RDD，repartition 操作都可以完成，因为无论如何都会经 shuffle 过程。

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

 def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2) //2个分区

    //repartition底层就是调用的coalesce 且增加了shuffle条件
    val newRDD: RDD[Int] = rdd.repartition(4) //扩大为4个分区

    newRDD.saveAsTextFile("output")
  //关闭环境
  sc.stop()
  

12、sortBy

        该操作用于排序数据。在排序之前，可以将数据通过 f 函数进行处理，之后按照 f 函数处理 的结果进行排序，默认为升序排列。排序后新产生的 RDD 的分区数与原 RDD 的分区数一 致。中间存在 shuffle 的过程

def sortBy[K](
 f: (T) => K,
ascending: Boolean = true,
 numPartitions: Int = this.partitions.length)
 (implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]

def main(args: Array[String]): Unit = 
    //准备环境
    //"*"代表线程的核数   应用程序名称"RDD"
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    val rdd: RDD[Int] = sc.makeRDD(List(2,1,6,5,9,4,8,7),2) //2个分区
    //sortBy(排序规则)
    val SRDD: RDD[Int] = rdd.sortBy(n=>n)
    //存在 shuffle 的过程
    SRDD.saveAsTextFile("output")

  //关闭环境
  sc.stop()