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 => Arrayval 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) //分区：2val 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()}