RDD算子

1. RDD算子介绍

RDD的算子，英文名为Operator, 可以理解为操作，其实操作就是编程代码中的方法。称呼的叫法源自认知心理学：
认知心理学认为解决问题其实将问题的状态进行改变:
问题(初始)=>操作(算子)=>问题(审核中)=>操作(算子)=> 问题(完成) Alt text 其中RDD方法就是RDD算子。为了区分Scala集合的操作方法，因为Scala的集合操作是在同一个节点内存中完成的，RDD的方法可以将计算逻辑发送到Excutor端(分布式节点执行)，于是将RDD的方法叫做算子，需要注意的是编程代码中：RDD的方法外部的操作都是在Driver端执行的，而方法内部的逻辑代码是在Excutor端执行的。

2. RDD转换算子

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

3. Value类型

3.1 map算子

函数签名: def map[U: ClassTag](f: T => U): RDD[U]
函数说明: 将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。

map算子的使用

scala

object RDD_Map {

    def main(args: Array[String]): Unit = {
        val sparkConf: 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))

        // 转换函数
        def mapFunction(num: Int): Unit = {
            num * 2
        }

        // val mapRDD: RDD[Unit] = rdd.map(mapFunction)
        // 简化，使用匿名函数
        rdd.map((num: Int) => {
            num * 2
        })
        // 再次简化，由于只有一行代码，可以省略大括号
        rdd.map((num: Int) => num * 2)
        // 再次简化，由于变量num也是Int类型, 可以使用上下文推断
        rdd.map((num) => num * 2)
        // 再次简化，由于括号中只有一个变量, 可以省略括号
        rdd.map(num => num * 2)
        // 再次简化，由于只有一个变量并且在方法体代码中按照顺序出现, 可以使用下划线_表示
        rdd.map(_ * 2)
        mapRDD.collect().foreach(println)
    }
}

map算子并行计算特点
- 一个分区内数据会依次一个一个的执行，只有前面一个数据全部的RDD计算完毕才会执行下一个数据的RDD计算
- 多个分区的数据执行是无序的

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 1)
    // 当前分区数
    println(rdd1.partitions.length)
    val rdd2: RDD[Int] = rdd1.map(num => {
        println(">>>>>>>" + num)
        num
    })

    val rdd3: RDD[Int] = rdd2.map((num) => {
        println("#######" + num)
        num
    })
    rdd3.saveAsTextFile("output/result")
    sc.stop()
}

运行结果：
Alt text 将分区数设置为2，重新执行查看结果：

scala

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

Alt text 因为3和1不在一个分区，两个分别都是分区中第一个数据，所以3、1分别优先4、2执行，但1和3是并行的，1和3的顺序是随机的。

3.2 mapPartitions算子

函数签名: def mapPartitions[U: ClassTag](f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false): RDD[U] 函数说明: 将待处理的数据以分区为单位发送到计算节点进行处理。

mapPartitions算子特点以分区为单位进行数据转换操作，但是会将整个分区的数据加载到内存进行引用。如果数据处理完的数据是不会释放掉，存在对象的引用。在内存较小，数据量较大的场合下，容易出现内存溢出。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 1)
    val rdd2: RDD[Int] = rdd1.mapPartitions(iterator => {
        println(">>>>>>>>>")
        iterator.map(_ * 2)
    })
    val rdd3: RDD[Int] = rdd2.mapPartitions(iterator => {
        println("#########")
        iterator.map(_ * 2)
    })
    rdd3.collect()
    sc.stop()
}

运行结果：
Alt text map和mapPartitions 的区别？

数据处理角度
Map算子是分区内一个数据一个数据的执行，类似于串行操作。
mapPartitions算子是以分区为单位进行批处理操作。
功能的角度
Map算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。
MapPartitions算子需要传递一个迭代器，返回一个迭代器，没有要求的元素的个数保持不变，所以可以增加或减少数据。
性能的角度
Map算子因为类似于串行操作，所以性能比较低， mapPartitions算子类似于批处理，所以性能较高。但是mapPartitions算子会长时间占用内存，那么这样会导致内存可能不够用，出现内存溢出的错误。所以在内存有限的情况下，不推荐使用。需要使用map算子，毕竟完成比完美更重要。

3.3 mapPartitionsWithIndex算子

函数签名: def mapPartitionsWithIndex[U: ClassTag](f: (Int, Iterator[T]) => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]
函数说明: 将待处理的数据以分区为单位发送到计算节点进行处理。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7), 3)
    // 过滤第一个分区
    val rdd2: RDD[Int] = rdd1.mapPartitionsWithIndex(
        (index, iterator)=>{
            if(index==0){
                Nil.iterator
            }else{
                iterator
            }
        }
    )
    // 将分区的数据变成元组,格式为(分区，元素)
    val rdd3 = rdd2.mapPartitionsWithIndex(
        (index, iterator) => {
            iterator.map(item =>{
                (index, item)
            })
        }
    )
    rdd3.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text

3.4 flatMap算子

函数签名：def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]
函数说明: 将处理的数据进行扁平化后再进行映射处理，所以算子也称之为扁平映射。比如：将 List(List(1,2),3,List(4,5))进行扁平化操作

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(List(1, 2, 3, 4), 5, List(6, 7)), 2)
    val rdd2: RDD[Int] = rdd1.flatMap(data => {
        // 匹配元素类型
        data match {
            case list: List[Int] => list  //如果是数组
            case num: Int => List(num)   // 如果是基本类型
        }
    })
    rdd2.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text

3.5 glom算子

函数签名: def glom(): RDD[Array[T]]
函数说明: 将同一个分区的数据直接转换为相同类型的内存数组进行处理，分区不变。
举例：计算所有分区最大值求和(分区内取最大值，分区间最大值求和)

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(1, 2, 3, 4,7),  2)

    /**
        * 转换成数组 
        * [1,2]  [3,4,7]
        * [2]  [7]
        * [9]
        */
    val rdd2: RDD[Array[Int]] = rdd1.glom()
    val rdd3: RDD[Int] = rdd2.map(array => {
        array.max
    })

    println(rdd3.collect().sum)
    sc.stop()
}

3.6 groupBy算子

函数签名：def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]
函数说明: 将数据根据指定的规则进行分组, 分区默认不变，但是数据会被打乱重新组合，我们将这样的操作称之为shuffle。极限情况下，数据可能被分在同一个分区中。一个组的数据在一个分区中，但是并不是说一个分区中只有一个组。
代码演练： List("Hello", "hive", "hbase", "Hadoop", "Spark", "Scala", "Jack")根据单词首写字母进行分组。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List("Hello", "hive", "hbase", "Hadoop", "Spark", "Scala", "Jack"),  2)
    // 根据首字母分组
    val rdd2: RDD[(Char, Iterable[String])] = rdd1.groupBy(it => {
        it.charAt(0)
    })
    rdd2.collect().foreach(println)
    println("分区数量：", rdd2.partitions.length)
    sc.stop()
}

运行结果：
Alt text 可以看到虽然分组有4个，但是分区仍然是2, 一个分区中可能不止一个组。

3.7 filter算子

函数签名：def filter(f: T => Boolean): RDD[T]
函数说明: 将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。当数据进行筛选过滤后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出现数据倾斜。

scala

val dataRDD = sparkContext.makeRDD(List(1,2,3,4),1)
val dataRDD1 = dataRDD.filter(_%2 == 0)

3.8 sample算子

函数签名：def sample(withReplacement: Boolean, fraction: Double, seed: Long = Utils.random.nextLong): RDD[T]
函数说明: sample来自古法语，有样品，样本的意思。sample算子表示根据指定的规则从数据集中抽取数据。其中withReplacement表示抽取后是否数据放回, seed表示随机算法种子，fraction看数据是否放回，如果不放回表示每条数据被抽到的概率，范围在[0,1]之间, 如果放回表示每条数据被抽到的次数, 范围大于等于0。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(0,1,2,3,4,5,6,7,8,9),  2)
    // 不放回数据进行取样
    val rdd2: RDD[Int] = rdd1.sample(false, 0.5, 1)
    println(rdd2.collect().mkString(","))
    sc.stop()
}

运行结果：
Alt text
可以看到结果为3个，需要注意的是概率为0.5不表示一定要抽5个, 具体底层是第三个参数如果固定下来，每个数据的概率就会被算出，然后取出比第二个参数(概率)大的数据, 所以多次执行程序结果不变。
如果不传第三个参数，默认使用当前时间作为种子：

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(0,1,2,3,4,5,6,7,8,9),  2)
    // 放回数据进行取样. 不传第三个参数，默认使用当前时间作为种子
    val rdd2: RDD[Int] = rdd1.sample(true, 0.5)
    println(rdd2.collect().mkString(","))
    sc.stop()
}

运行2次结果： Alt text 查看sample算子源码：

scala

def sample(
    withReplacement: Boolean,
    fraction: Double,
    seed: Long = Utils.random.nextLong): RDD[T] = {
require(fraction >= 0,
    s"Fraction must be nonnegative, but got ${fraction}")

withScope {
    if (withReplacement) {
    // 抽取数据放回(泊松算法)      
    new PartitionwiseSampledRDD[T, T](this, new PoissonSampler[T](fraction), true, seed)
    } else {
    //  抽取数据不放回(伯努利算法)又叫 0、1 分布。例如扔硬币，要么正面，要么反面。
    new PartitionwiseSampledRDD[T, T](this, new BernoulliSampler[T](fraction), true, seed)
    }
}
}

应用场景：检测数据倾斜是否发生，执行sample算子进行随机取样。

3.9 distinct算子

函数签名：
def distinct()(implicit ord: Ordering[T] = null): RDD[T]
def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]
函数说明: 将数据集中重复的数据去重。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(1,2,3,4,5,3,2,1,4),  2)
    val rdd2: RDD[Int] = rdd1.distinct()
    println(rdd2.collect().mkString(","))
    sc.stop()
}

Alt text

3.10 coalesce算子

函数签名： def coalesce(numPartitions: Int, shuffle: Boolean = false, partitionCoalescer: Option[PartitionCoalescer] = Option.empty) (implicit ord: Ordering[T] = null): RDD[T]
函数说明: 根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率当Spark程序中，存在过多的小任务的时候，可以通过coalesce方法，收缩合并分区，减少分区的个数，减小任务调度成本。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(1,2,3,4,5,3,2,1,4),  5)
    // 缩减分区, 将每个被缩减分区里面所有元素转移到其他一个分区中，而不会将分区中的数据重新打乱组合到多个分区。
    //容易导致数据倾斜，需要数据均衡，需要设置shuffle=true
    // val rdd2: RDD[Int] = rdd1.coalesce(3)
    val rdd2: RDD[Int] = rdd1.coalesce(3, true)
    rdd2.saveAsTextFile("output/result")
    sc.stop()
}

运行结果： Alt text

3.11 repartition算子

函数签名: def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]
函数说明: 该操作内部其实执行的是coalesce操作，参数shuffle的默认值为true。无论是将分区数多的RDD转换为分区数少的RDD，还是将分区数少的 RDD 转换为分区数多的RDD，repartition操作都可以完成，因为无论如何都会经shuffle过程。

scala

val dataRDD = sparkContext.makeRDD(List(1,2,3,4,1,2),2)
val dataRDD1 = dataRDD.repartition(4)

提示

coalesce算子和repartition算子可以用来合并小文件，提交计算效率。

3.12 sortBy算子

函数签名: def sortBy[K]( f: (T) => K,ascending: Boolean = true,numPartitions: Int = this.partitions.length)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]
函数说明:该操作用于排序数据。在排序之前可以将数据通过f函数进行处理，之后按照f函数处理的结果进行排序，默认为升序排列。排序后新产生的RDD的分区数与原 RDD的分区数一致，中间存在shuffle的过程。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(1, 2, 3, 4, 5, 3, 2, 1, 4), 2)
    val rdd2: RDD[Int] = rdd1.sortBy(num => num)
    rdd2.saveAsTextFile("output/result")
    sc.stop()
}

Alt text

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1: RDD[(String, String)] = sc.makeRDD(List(("1", "jack"), ("11", "demo"), ("2", "test")))
    // 按照指定规则排序，默认为升序，第二个参数可以改变顺序
    val rdd2: RDD[(String, String)] = rdd1.sortBy(t => t._1.toInt, false)
    rdd2.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text

4. 双Value类型

4.1 intersection算子

函数签名: def intersection(other: RDD[T]): RDD[T]
函数说明: 对源RDD和参数RDD求交集后返回一个新的RDD

scala

val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4))
val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6))
val dataRDD = dataRDD1.intersection(dataRDD2)
println(dataRDD.collect().mkString(","))

执行结果： Alt text ❓如果两个RDD数据类型不一致怎么办？
编译不通过。查看源码：

4.2 union算子

函数签名: def union(other: RDD[T]): RDD[T]
函数说明: 对源RDD和参数RDD求并集后返回一个新的RDD

scala

val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4))
val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6))
val dataRDD = dataRDD1.union(dataRDD2)

❓如果两个RDD数据类型不一致怎么办？
编译不通过。查看源码：
Alt text

4.3 subtract算子

函数签名: def subtract(other: RDD[T]): RDD[T] 函数说明: 以一个 RDD 元素为主，去除两个 RDD 中重复元素，将其他元素保留下来。求差集

scala

val dataRDD1 = sparkContext.makeRDD(List(1,2,3,4))
val dataRDD2 = sparkContext.makeRDD(List(3,4,5,6))
val dataRDD = dataRDD1.subtract(dataRDD2)

❓如果两个RDD数据类型不一致怎么办？
编译不通过。查看源码：
Alt text

4.4 zip算子

函数签名: def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)] 函数说明: 将两个 RDD 中的元素，以键值对的形式进行合并。其中键值对中的Key为第1个RDD中的元素，Value为第2个RDD中的相同位置的元素。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val dataRDD1 = sc.makeRDD(List(1, 2, 3, 4))
    val dataRDD2 = sc.makeRDD(List(3, 4, 5, 6))
    val dataRDD = dataRDD1.zip(dataRDD2)
    dataRDD.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text zip算子支持两个RDD数据源类型不一致。但是要求两个RDD具有相同的分区和分区中数据个数相同。

5. Key-Value类型

5.1 partitionBy算子

函数签名: def partitionBy(partitioner: Partitioner): RDD[(K, V)] 函数说明: 将数据按照指定Partitioner重新进行分区。Spark默认的分区器是HashPartitioner。
Alt text

提示

partitionBy算子和coalesce算子、repartition算子不同，前者不改变分区数量，只进行调整分区中的数据在哪个分区。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1: RDD[Int] = sc.makeRDD(List(5, 2, 3, 6, 7, 4, 9), 2)
    //  转换成元组，记录分区信息
    val rdd2: RDD[(Int, Int)] = rdd1.mapPartitionsWithIndex((index, it) => {
        val iterator: Iterator[(Int, Int)] = it.map(item => (item, index))
        iterator
    })
    // 重新组合，利用hash分区器
    val rdd3: RDD[(Int, Int)] = rdd2.partitionBy(new HashPartitioner(2))
    rdd3.saveAsTextFile("output/result/")
    sc.stop()
}

运行结果：
Alt text ❓如果重分区的分区器和当前RDD的分区器一样怎么办？
也就是这个意思：

scala

val rdd3: RDD[(Int, Int)] = rdd2.partitionBy(new HashPartitioner(2))
// 重分区
val rdd4: RDD[(Int, Int)] = rdd3.partitionBy(new HashPartitioner(2))

分区结果和之前的一致没有变化，可以查看源码：
partitionBy方法源码:
partitionBy源码图 HashPartitioner分区器源码:
HashPartitioner分区器源码图

5.2 reduceByKey算子

函数签名：
def reduceByKey(func: (V, V) => V): RDD[(K, V)]
def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
函数说明: 可以将数据按照相同的Key对Value进行聚合运算。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a",1), ("a", 5), ("a", 8), ("b", 4)),  2)
    val rdd2: RDD[(String, Int)] = rdd1.reduceByKey((value1, value2) => {
        println(s"$value1 + $value2")
        value1 + value2
    })
    rdd2.collect().foreach(println)
    sc.stop()
}

运行结果：
reduceByKey算子代码示例可以看到reduceByKey中key的数据只有一个，是不会参与运算的。

5.3 groupByKey算子

函数签名:
def groupByKey(): RDD[(K, Iterable[V])]
def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]
def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]
函数说明: 将数据源的数据根据key对value进行分组, 分组中第一个元素时key, 第二个元素是相同key的value集合。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a", 1), ("a", 3), ("a", "6"), ("b", 4)),  2)
    val rdd2: RDD[(String, Iterable[Any])] = rdd1.groupByKey()
    rdd2.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text ❓reduceByKey和groupByKey的区别？
groupByKey的工作流程如下：

reduceByKey的工作流程如下：总结：

从shuffle的角度：reduceByKey和groupByKey都存在shuffle的操作，但是reduceByKey可以在shuffle前对分区内相同key的数据进行预聚合(combine)功能，这样会减少落盘的数据量，而groupByKey只是进行分组，不存在数据量减少的问题，reduceByKey性能比较高。
从功能的角度：reduceByKey其实包含分组和聚合的功能。groupByKey只能分组，不能聚合，所以在分组聚合的场合下，推荐使用reduceByKey，如果仅仅是分组而不需要聚合。那么使用groupByKey。

5.4 aggregateByKey算子

函数签名:
def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)]
函数说明: 将数据根据不同的规则进行分区内计算和分区间计算。其中zeroValue表示初始值。
需求：取出每个分区内相同 key 的最大值然后分区间相加

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a",1), ("a",2), ("b",3), ("b",4), ("b",5), ("a",6)),  2)
    // aggregateByKey算子是函数柯里化，存在两个参数列表
    // 1. 第一个参数列表中的参数表示初始值
    // 2. 第二个参数列表中含有两个参数
    // 2.1 第一个参数表示分区内的计算规则
    // 2.2 第二个参数表示分区间的计算规则
    val rdd2: RDD[(String, Int)] = rdd1.aggregateByKey(0)(
        (value1, value2) => math.max(value1, value2),
        (x, y) => x + y
    )
    rdd2.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text 设置初始值的原因是，aggregateByKey在分区是从第一个数据开始的，计算需要两个值, 初始值被用来参与第一次计算。
运行流程：
aggregateByKey最终返回结果应该和初始值的类型保持一致，如下示例：

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2), ("a", 6), ("b", 8), ("a", 2)),2)
    // 求平均值
    // 初始值为一个元组, 元组的第一个元素为和的初始值0，第二个元素为累加次数，也就是相同key的元素数量
    val rdd2: RDD[(String, (Int, Int))] = rdd1.aggregateByKey((0, 0))(
        // 传递参数第一个就是元组类型，并作为结果在二次计算传入
        (t, v) => {
            (t._1 + v, t._2 + 1)  // 第二个元素每次计数累加
        },
        // 分区间的数据计算规则
        (t1, t2) => {
            (t1._1 + t2._1, t1._2 + t2._2)
        }
    )
    val rdd3: RDD[(String, Int)] = rdd2.mapValues(t1 => {
        t1._1 / t1._2
    })
    rdd3.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text

5.5 foldByKey算子

函数签名：def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]
函数说明: 当分区内计算规则和分区间计算规则相同时，aggregateByKey就可以简化为foldByKey。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a",1), ("a",2), ("b",3), ("b",4), ("b",5), ("a",6)),  2)
    // aggregateByKey算子是函数柯里化，存在两个参数列表
    // 1. 第一个参数列表中的参数表示初始值
    // 2. 第二个参数列表中含有一个参数
    // 2.1 一个参数表示分区内的计算规则
    val rdd2: RDD[(String, Int)] = rdd1.foldByKey(0)(
        (x, y) => x + y
    )
    rdd2.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text

5.6 combineByKey算子

函数签名: def combineByKey[C](createCombiner: V => C,mergeValue: (C, V) => C,mergeCombiners: (C, C) => C): RDD[(K, C)]
函数说明: 最通用的对key-value型Rdd进行聚集操作的聚集函数(aggregation function)。类似于aggregate(),combineByKey()允许用户返回值的类型与输入不一致。它允许对计算的第一个数据进行格式转换。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)), 2)
    // 第一个参数表示： 将相同key的第一个数据进行结构转换
    // 第二个参数表示： 分区内的计算规则
    // 第三个参数表示： 分区外的计算规则
    val rdd2: RDD[(String, (Int, Int))] = rdd1.combineByKey(
        v => (v, 1),
        (t: (Int, Int), v) => {
            (t._1 + v, t._2 + 1)
        },
        (t1: (Int, Int), t2: (Int, Int)) => {
            (t1._1 + t2._1, t1._2 + t1._2)
        }
    )
    val rdd3: RDD[(String, Int)] = rdd2.mapValues(v => {
        v._1 / v._2
    })
    rdd3.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text

总结：reduceByKey、foldByKey、aggregateByKey、combineByKey的区别 reduceByKey: 相同key的第一个数据不进行任何计算，分区内和分区间计算规则相同。
foldByKey: 相同key的第一个数据和初始值进行分区内计算，分区内和分区间计算规则相同。
aggregateByKey：相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则可以不相同。
combineByKey:当计算时，发现数据结构不满足要求时，可以让第一个数据转换结构。分区内和分区间计算规则不相同。他们的底层都是调用的combineByKeyWithClassTag函数。

5.7 sortByKey算子

函数签名: def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)] 函数说明: 在一个(K,V)的 RDD上调用，K必须实现Ordered接口(特质)，返回一个按照key进行排序的

scala

val dataRDD1 = sparkContext.makeRDD(List(("a",1),("b",2),("c",3)))
val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(true)
val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(false)

5.8 join算子

函数签名: def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))] 函数说明: 在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素连接在一起的(K,(V,W))的RDD

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a", 1), ("c", 2), ("b", 3)), 2)
    val rdd2 = sc.makeRDD(List(("a", 2), ("b", 4), ("c", 5)), 2)
    /**
        * 两个不同数据源的数据，相同的key的value会连接在一起，形成元组
        * 如果两个数据源中的key没有匹配上，那么数据不会出现在结果中
        * 如果两个数据源中key有多个相同的，会依次匹配，可能会出现笛卡尔积，数据几何倍增长。
        */
    val rdd3: RDD[(String, (Int, Int))] = rdd1.join(rdd2)
    rdd3.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text

提示

join算子谨慎使用，最好能用其他算子替代使用实现。

5.9 leftOuterJoin算子

函数签名： def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
函数说明：类似于SQL语句的左外连接

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a", 1), ("c", 2), ("b", 3)), 2)
    val rdd2 = sc.makeRDD(List(("a", 2), ("b", 4)), 2)

    val rdd3: RDD[(String, (Int, Option[Int]))] = rdd1.leftOuterJoin(rdd2)
    rdd3.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text 其中None类似Java8中Option, 表示为空。

5.10 rightOuterJoin算子

函数签名： def rightOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
函数说明：类似于SQL语句的右外连接

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a", 1), ("c", 2), ("b", 3)), 2)
    val rdd2 = sc.makeRDD(List(("a", 2), ("b", 4)), 2)

    val rdd3: RDD[(String, (Option[Int], Int))] = rdd1.rightOuterJoin(rdd2)
    rdd3.collect().foreach(println)
    sc.stop()
}

运行结果： Alt text

5.11 cogroup算子

函数签名： def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))] 函数说明: 在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a", 1), ("c", 2), ("b", 3), ("a", 10)), 2)
    val rdd2 = sc.makeRDD(List(("a", 2), ("b", 4)), 2)

    // 分组 连接
    val rdd3: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd1.cogroup(rdd2)
    rdd3.collect().foreach(println)
    sc.stop()
}

运行结果：
Alt text

6. 行动算子

他们都有个特点，立即执行，并且返回结果而不是RDD。查看Spark源码：

scala

......
// Spark上下文调用任务执行
val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
......
}

6.1 reduce算子

函数签名：def reduce(f: (T, T) => T): T
函数说明: 聚集RDD中的所有元素，先聚合分区内数据，再聚合分区间数据

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(1,2,3,4),  2)
    val result: Int = rdd1.reduce(_ + _)
    println(s"执行结果：$result")
    sc.stop()
}

执行结果：
Alt text

6.2 collect算子

函数签名：def collect(): Array[T]
函数说明: 在驱动程序中，以数组Array的形式返回数据集的所有元素。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(1,2,3,4),  2)
    // 方法会将不同分区的数据按照分区的顺序采集到Driver端内存中，形成数组
    val ints: Array[Int] = rdd1.collect()
    println(ints.mkString(","))
    sc.stop()
}

执行结果：
Alt text

6.3 count算子

函数签名: def count(): Long
函数说明: 返回RDD中元素的个数

scala

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
// 返回RDD中元素的个数
val countResult: Long = rdd.count()

6.4 first算子

函数签名: first(): T 函数说明: 返回RDD中的第一个元素

scala

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
// 返回RDD中第一个元素
val firstResult: Int = rdd.first()
println(firstResult)

6.5 take算子

函数签名: def take(num: Int): Array[T]
函数说明: 返回一个由RDD的前n个元素组成的数组

scala

val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4))
// 获取N个数据
val takeResult: Array[Int] = rdd.take(2)
println(takeResult.mkString(","))

6.6 takeOrdered算子

函数签名: def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]
函数说明: 返回该RDD排序后的前n个元素组成的数组

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(1,3,5,4,2),  2)
    // 自定义排序规则，按照元素的倒序排序, 
    // takeOrdered方法用了隐式参数, 编译器会查找作用域内的隐式值descendingOrder来填充ord参数
    implicit val descendingOrder = Ordering.Int.reverse
    val ints: Array[Int] = rdd1.takeOrdered(3)
    println(ints.mkString(","))
    sc.stop()
}

运行结果：
Alt text

6.7 aggregate算子

函数签名: def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U
函数说明: 分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的数据聚合。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(1,3,5,4,2),  2)
    // aggregateByKey：初始值只会参与分区内的计算
    // aggregate: 初始值在分区内和分区间的计算都会参与
    // ((1+3+10)+(5+4+2+10)+10) = 45
    val result: Int = rdd1.aggregate(10)(_ + _, _ + _)
    println(s"执行结果：$result")
    sc.stop()
}

运行结果：
Alt text

6.8 fold算子

函数签名: def fold(zeroValue: T)(op: (T, T) => T): T
函数说明: 分区的数据通过初始值和分区内的数据进行聚合，分区内和分区间的计算规则相同，aggregate的简化版操作。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(1,3,5,4,2),  2)
    
    val result: Int = rdd1.fold(10)(_ + _)
    println(s"执行结果：$result")
    sc.stop()
}

运行结果：
Alt text

6.8 countByKey算子

函数签名: def countByKey(): Map[K, Long]
函数说明: 统计每种key的个数

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a", 1),("b", 1),("a", 3),("a", 4)),  2)
    val result: collection.Map[String, Long] = rdd1.countByKey()
    println(s"执行结果：$result")
    sc.stop()
}

运行结果：
Alt text

6.8 countByValue算子

函数签名: def countByValue()(implicit ord: Ordering[T] = null): Map[T, Long]
函数说明: 统计每种Value的个数,不同于countByKey，countByValue用于Map类型数组也可以用于简单类型数组

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a", 1),("b", 1),("a", 3),("a", 4)),  2)
    val result: collection.Map[(String, Int), Long] = rdd1.countByValue()
    println(s"执行结果：$result")
    sc.stop()
}

运行结果：
Alt text

6.8 save相关算子

函数签名:
def saveAsTextFile(path: String): Unit
def saveAsObjectFile(path: String): Unit
def saveAsSequenceFile(path: String,codec: Option[Class[_ <: CompressionCodec]] = None): Unit
函数说明: 将数据保存到不同格式的文件中

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List(("a", 1),("b", 1),("a", 3),("a", 4)),  2)
    rdd1.saveAsTextFile("output/result1")
    rdd1.saveAsObjectFile("output/result2")
    // 要求数据必须是k-v类型
    rdd1.saveAsSequenceFile("output/result3")
    sc.stop()
}

运行结果：
Alt text

6.8 foreach算子

函数签名:
def foreach(f: T => Unit): Unit = withScope { val cleanF = sc.clean(f) sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))}
函数说明: 分布式遍历RDD中的每一个元素，调用指定函数。

scala

def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    val rdd1 = sc.makeRDD(List( 1, 3, 4, 2),  2)
    // foreach是Driver端内存集合遍历打印
    rdd1.collect().foreach(println)
    println("**********************")
    // 其实是Executor端内存数据打印
    rdd1.foreach(println)
    sc.stop()
}

运行结果：
Alt text 可以发现两次打印结果不一样，原因在于collect()算子已经将数据返回给Driver端，这时数据打印是有顺序的，foreach不是算子而是Scala的foreach函数，而第二个foreach是算子，在Excutor端(分布式节点)中的RDD内部执行打印, 分布式节点执行是并行计算，打印结果顺序不可控。
打印流程图

RDD算子 ​

1. RDD算子介绍 ​

2. RDD转换算子 ​

3. Value类型 ​

3.1 map算子 ​

3.2 mapPartitions算子 ​

3.3 mapPartitionsWithIndex算子 ​

3.4 flatMap算子 ​

3.5 glom算子 ​

3.6 groupBy算子 ​

3.7 filter算子 ​

3.8 sample算子 ​

3.9 distinct算子 ​

3.10 coalesce算子 ​

3.11 repartition算子 ​

3.12 sortBy算子 ​

4. 双Value类型 ​

4.1 intersection算子 ​

4.2 union算子 ​

4.3 subtract算子 ​

4.4 zip算子 ​

5. Key-Value类型 ​

5.1 partitionBy算子 ​

5.2 reduceByKey算子 ​

5.3 groupByKey算子 ​

5.4 aggregateByKey算子 ​

5.5 foldByKey算子 ​

5.6 combineByKey算子 ​

5.7 sortByKey算子 ​

5.8 join算子 ​

5.9 leftOuterJoin算子 ​

5.10 rightOuterJoin算子 ​

5.11 cogroup算子 ​

6. 行动算子 ​

6.1 reduce算子 ​

6.2 collect算子 ​

6.3 count算子 ​

6.4 first算子 ​

6.5 take算子 ​

6.6 takeOrdered算子 ​

6.7 aggregate算子 ​

6.8 fold算子 ​

6.8 countByKey算子 ​

6.8 countByValue算子 ​

6.8 save相关算子 ​

6.8 foreach算子 ​

RDD算子

1. RDD算子介绍

2. RDD转换算子

3. Value类型

3.1 map算子

3.2 mapPartitions算子

3.3 mapPartitionsWithIndex算子

3.4 flatMap算子

3.5 glom算子

3.6 groupBy算子

3.7 filter算子

3.8 sample算子

3.9 distinct算子

3.10 coalesce算子

3.11 repartition算子

3.12 sortBy算子

4. 双Value类型

4.1 intersection算子

4.2 union算子

4.3 subtract算子

4.4 zip算子

5. Key-Value类型

5.1 partitionBy算子

5.2 reduceByKey算子

5.3 groupByKey算子

5.4 aggregateByKey算子

5.5 foldByKey算子

5.6 combineByKey算子

5.7 sortByKey算子

5.8 join算子

5.9 leftOuterJoin算子

5.10 rightOuterJoin算子

5.11 cogroup算子

6. 行动算子

6.1 reduce算子

6.2 collect算子

6.3 count算子

6.4 first算子

6.5 take算子

6.6 takeOrdered算子

6.7 aggregate算子

6.8 fold算子

6.8 countByKey算子

6.8 countByValue算子

6.8 save相关算子

6.8 foreach算子