2022-03-24 | 标签 Spark

1. 什么是RDD

RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark 中最基本的数据处理模型。代码中是一个抽象类，它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。

弹性

存储的弹性：内存与磁盘的自动切换；容错的弹性：数据丢失可以自动恢复；计算的弹性：计算出错重试机制；分片的弹性：可根据需要重新分片。

分布式：数据存储在大数据集群不同节点上
数据集：RDD 封装了计算逻辑，并不保存数据
数据抽象：RDD 是一个抽象类，需要子类具体实现
不可变：RDD 封装了计算逻辑，是不可以改变的，想要改变，只能产生新的RDD，在新的RDD 里面封装计算逻辑
可分区、并行计算

2. 核心属性

分区列表

RDD 数据结构中存在分区列表，用于执行任务时并行计算，是实现分布式计算的重要属性。

分区计算函数

Spark 在计算时，是使用分区函数对每一个分区进行计算

RDD 之间的依赖关系

RDD 是计算模型的封装，当需求中需要将多个计算模型进行组合时，就需要将多个RDD 建立依赖关系

分区器（可选）

当数据为KV 类型数据时，可以通过设定分区器自定义数据的分区

首选位置（可选）

计算数据时，可以根据计算节点的状态选择不同的节点位置进行计算

3. 执行原理

从计算的角度来讲，数据处理过程中需要计算资源（内存& CPU）和计算模型（逻辑）。执行时，需要将计算资源和计算模型进行协调和整合。

Spark 框架在执行时，先申请资源，然后将应用程序的数据处理逻辑分解成一个一个的计算任务。然后将任务发到已经分配资源的计算节点上, 按照指定的计算模型进行数据计算。最后得到计算结果。

RDD 是Spark 框架中用于数据处理的核心模型，接下来我们看看，在Yarn 环境中，RDD 的工作原理:

启动Yarn 集群环境

Spark 通过申请资源创建调度节点和计算节点

Spark 框架根据需求将计算逻辑根据分区划分成不同的任务任务放到任务里，需要被调度

调度节点将任务根据计算节点状态发送到对应的计算节点进行计算

从以上流程可以看出RDD 在整个流程中主要用于将逻辑进行封装，并生成Task 发送给Executor 节点执行计算，接下来我们就一起看看Spark 框架中RDD 是具体是如何进行数据处理的。

4. 基础编程

4.1 RDD创建

在Spark 中创建RDD 的创建方式可以分为四种：

4.1.1 从集合（内存）中创建RDD

从集合中创建RDD，Spark主要提供了两个方法：parallelize 和makeRDD

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark") 
val sparkContext = new SparkContext(sparkConf)
val rdd1 = sparkContext.parallelize( List(1,2,3,4))
val rdd2 = sparkContext.makeRDD( List(1,2,3,4))

从底层代码实现来讲，makeRDD 方法其实就是parallelize 方法

def makeRDD[T: ClassTag](
      seq: Seq[T],
      numSlices: Int = defaultParallelism): RDD[T] = withScope {
    parallelize(seq, numSlices)
  }

4.1.2 从外部存储（文件）创建RDD

由外部存储系统的数据集创建RDD 包括：本地的文件系统，所有Hadoop 支持的数据集，比如HDFS、HBase 等。

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark") 
val sparkContext = new SparkContext(sparkConf)
val fileRDD: RDD[String] = sparkContext.textFile("input") 
fileRDD.collect().foreach(println)
sparkContext.stop()

4.1.3 从其他RDD 创建

主要是通过一个RDD 运算完后，再产生新的RDD。

4.1.4 直接创建RDD（new）

使用new 的方式直接构造RDD，一般由Spark 框架自身使用。

4.2 RDD 并行度与分区

默认情况下，Spark 可以将一个作业切分多个任务后，发送给Executor 节点并行计算，而能够并行计算的任务数量我们称之为并行度。这个数量可以在构建RDD 时指定。记住，这里的并行执行的任务数量，并不是指的切分任务的数量，不要混淆了。

val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark") 
val sparkContext = new SparkContext(sparkConf)
val dataRDD: RDD[Int] = sparkContext.makeRDD(List(1,2,3,4), 4)
val fileRDD: RDD[String] = sparkContext.textFile("input", 2)
fileRDD.collect().foreach(println) 
sparkContext.stop()

读取内存数据时，数据可以按照并行度的设定进行数据的分区操作，数据分区规则的

Spark 核心源码如下：

def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = 
{ (0 until numSlices).iterator.map { i =>
    val start = ((i * length) / numSlices).toInt
    val end = (((i + 1) * length) / numSlices).toInt 
  (start, end)
}
}

读取文件数据时，数据是按照Hadoop 文件读取的规则进行切片分区，而切片规则和数据读取的规则有些差异，具体Spark 核心源码如下 ```scala public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException { long totalSize = 0; for (FileStatus file: files) { if (file.isDirectory()) { throw new IOException(“Not a file: “+ file.getPath()); } totalSize += file.getLen(); }

long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits); long minSize = Math.max(job.getLong(org.apache.hadoop.mapreduce.lib.input. FileInputFormat.SPLIT_MINSIZE, 1), minSplitSize);

…

for (FileStatus file: files) {

…

if (isSplitable(fs, path)) { long blockSize = file.getBlockSize(); long splitSize = computeSplitSize(goalSize, minSize, blockSize);

…

} protected long computeSplitSize(long goalSize, long minSize,long blockSize) return Math.max(minSize, Math.min(goalSize, blockSize)); }

### 
## 4.3 RDD 转换算子
RDD 根据数据处理方式的不同将算子整体上分为Value 类型、双Value 类型和Key-Value类型
### 4.3.1 Value类型
#### 1. map
函数签名
def map[U: ClassTag](f: T => U): RDD[U]

函数说明
将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。
```scala
val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparkConf)

    //TODO 算子 - map
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
    // 1,2,3,4
    // 2,4,6,8

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

//    val mapRDD: RDD[Int] = rdd.map(mapFunction)

//    val mapRDD: RDD[Int] = rdd.map((num: Int) => {num*2})
//    val mapRDD: RDD[Int] = rdd.map(num => num*2)
    val mapRDD: RDD[Int] = rdd.map( _ * 2 )

    mapRDD.collect().foreach( println )

    sc.stop()

####

2. mapPartitions

函数签名 def mapPartitionsU: ClassTag: RDD[U]

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

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

    //mapPartitions: 可以以分区为单位进行数据转换操作
    //               但是会将整个分区的数据加载到内存进行引用
    //               如果处理完数据是不会被释放掉的，因为存在对象的引用
    //               在内存较小，数据量较大的场合下，容易出现内存溢出
    val mpRDD: RDD[Int] = rdd.mapPartitions(
      iter => {
        println(">>>>>>>>>>>>>>>>>>")
        iter.map(_ * 2)
      }
    )

思考一个问题：map 和mapPartitions 的区别？

数据处理角度

Map 算子是分区内一个数据一个数据的执行，类似于串行操作。而mapPartitions 算子是以分区为单位进行批处理操作。

功能的角度

Map 算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。MapPartitions 算子需要传递一个迭代器，返回一个迭代器，没有要求的元素的个数保持不变，所以可以增加或减少数据

性能的角度

Map 算子因为类似于串行操作，所以性能比较低，而是mapPartitions 算子类似于批处理，所以性能较高。但是mapPartitions 算子会长时间占用内存，那么这样会导致内存可能不够用，出现内存溢出的错误。所以在内存有限的情况下，不推荐使用。使用map 操作。 完成比完美更重要

3. mapPartitionsWithIndex

函数签名 def mapPartitionsWithIndexU: ClassTag: RDD[U]

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

// 获取第二个分区的数据
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)
    //[1,2]  [3,4]
    val mpiRDD: RDD[Int] = rdd.mapPartitionsWithIndex(
      (index, iter) => {
        if (index == 1) {
          iter
        } else {
          Nil.iterator
        }
      }
    )

3
4

4. flatMap

函数签名 def flatMapU: ClassTag: RDD[U]

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

  val rdd = sc.makeRDD(List(List(1,2),3,List(4,5)))

    val flatRDD = rdd.flatMap(
      data => {
        data match {
          case list: List[_] => list
          case dat => List(dat)
        }
      }
    )

5. glom

函数签名 def glom(): RDD[Array[T]]

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

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

    //List => Int : flatMap
    //List => Array : glom

    val glomRDD: RDD[Array[Int]] = rdd.glom()

    glomRDD.collect().foreach(data => println(data.mkString(",")))

1,2
3,4,5

####

6. groupBy

函数签名 def groupByK(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]

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

val rdd: RDD[String] = sc.makeRDD(List("Hello", "hive", "hbase", "Hadoop"),2)

    // groupBy会将数据源中的每一个数据进行分组判断，根据返回的分组key进行分组
    // 相同的key值的数据会放置在一个组中

    //分组和分区没有必然的关系
    val groupRDD: RDD[(Char, Iterable[String])] = rdd.groupBy(_.charAt(0))

    groupRDD.collect().foreach( println )

(h,CompactBuffer(hive, hbase))
(H,CompactBuffer(Hello, Hadoop))

7. filter

函数签名 def filter(f: T => Boolean): RDD[T]

函数说明将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。当数据进行筛选过滤后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出现数据倾斜。

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

 val filterRDD: RDD[Int] = rdd.filter(num => num % 2 != 0)

 filterRDD.collect().foreach( println )

1
3

8. sample

函数签名 def sample( withReplacement: Boolean, fraction: Double, seed: Long = Utils.random.nextLong): RDD[T]

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

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

    //sample算子需要传递三个参数
    //1.第一个参数表示，抽取数据后是否将数据放回  true（放回），false（不放回）
    // 抽取数据不放回（伯努利算法）
    // 伯努利算法：又叫 0、1 分布。例如扔硬币，要么正面，要么反面。
    // 抽取数据放回（泊松算法）
    //2.第二个参数表示，
    //        如果抽取不放回的场合：数据源中每条数据被抽取的概率
    //        如果抽取放回的场合：表示数据源中的每条数据被抽取的可能次数
    //3.第三个参数表示，抽取数据时随机算法的种子
    //种子确定好，随机数也被确定了
    //如果不传递第三个参数，那么使用的就是当前系统时间

    println(rdd.sample(
      false,
      0.4,
      //1
    ).collect().mkString(","))

    println("==========================")

    println(rdd.sample(
      true,
      3,
      //1
    ).collect().mkString(","))

1,2,3,5,7,9
==========================
2,2,3,3,4,4,5,5,5,5,5,6,7,7,8,8,9,9,10,10,10,10,10

9. distinct

函数签名 def distinct()(implicit ord: Ordering[T] = null): RDD[T]

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

函数说明将数据集中重复的数据去重

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

    val data: RDD[Int] = rdd.distinct()
    //底层源码逻辑
    //map(x => (x, null)).reduceByKey((x, _) => x, numPartitions).map(_._1)

    //(1,null), (2,null), (3,null), (4,null), (1,null), (2,null), (3,null), (4,null)
    //(1,null), (2,null), (3,null), (4,null)
    //1, 2, 3, 4

    data.collect().foreach(println)

####

10. coalesce

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

函数说明根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率

当spark 程序中，存在过多的小任务的时候，可以通过coalesce 方法，收缩合并分区，减少分区的个数，减小任务调度成本

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

    //coalesce默认不会将分区的数据打乱重新组合
    //这种情况下的数据缩减分区可能会导致数据不均衡，导致数据倾斜
    //val newRDD: RDD[Int] = rdd.coalesce(2)
    //[1,2]
    //[3,4,5,6]

    //如果想要让数据均衡，可以进行shuffle处理
    val newRDD: RDD[Int] = rdd.coalesce(2,true)
    //[1,4,5]
    //[2,3,6]

11. repartition

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

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

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

    //coalesce算子可以扩大分区，但是如果不进行shuffle操作，是没有意义的，不起作用
    //val newRDD: RDD[Int] = rdd.coalesce(3)

    //所以如果想要实现扩大分区的效果，需要使用shuffle操作
    //val newRDD: RDD[Int] = rdd.coalesce(3,true)
    //[3,5]  [1,6]  [2,4]

    //spark提供了一个简化的操作
    //缩减分区：coalesce，如果想要数据均衡，可以采用shuffle
    //扩大分区：repartition
    val newRDD: RDD[Int] = rdd.repartition(3)
    //[3,5]  [1,6]  [2,4]
    //底层源码
    /*
    def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
      coalesce(numPartitions, shuffle = true)
    }
     */

12 sortBy

函数签名 def sortByK (implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]

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

    val rdd = sc.makeRDD(List(("1",1), ("11",2), ("2",3)), 2)

    val sortRDD: RDD[(String, Int)] = rdd.sortBy(t => t._1)
    //(1,1)
    //(11,2)
    //(2,3)

    val sortRDD1: RDD[(String, Int)] = rdd.sortBy(t => t._1.toInt)
    //(1,1)
    //(2,3)
    //(11,2)

    //sortBy方法可以根据指定的规则对数据源中的数据进行排序，默认是升序，第二个参数可以改变排序的方式
    //sortBy默认情况下，不会改变分区，但是中间存在shuffle操作
    val sortRDD2: RDD[(String, Int)] = rdd.sortBy(t => t._1.toInt,false)
    //(11,2)
    //(2,3)
    //(1,1)

4.3.2 双Value类型

1. intersection

函数签名 def intersection(other: RDD[T]): RDD[T]

函数说明对源RDD 和参数RDD 求交集后返回一个新的RDD

2. union

函数签名 def union(other: RDD[T]): RDD[T]

函数说明对源RDD 和参数RDD 求并集后返回一个新的RDD

3. subtract

函数签名 def subtract(other: RDD[T]): RDD[T]

函数说明以一个RDD 元素为主，去除两个RDD 中重复元素，将其他元素保留下来。求差集

4. zip

函数签名 def zipU: ClassTag: RDD[(T, U)]

函数说明将两个RDD 中的元素，以键值对的形式进行合并。其中，键值对中的Key 为第1 个RDD中的元素，Value 为第2 个RDD 中的相同位置的元素

    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
    val rdd2: RDD[Int] = sc.makeRDD(List(3, 4, 5, 6))
    val rdd3 = sc.makeRDD(List("3", "4", "5", "6"))

    //交集
    val interRDD: RDD[Int] = rdd1.intersection(rdd2)
    println(interRDD.collect().mkString(","))
    println("============================")

    //并集
    val unionRDD: RDD[Int] = rdd1.union(rdd2)
    println(unionRDD.collect().mkString(","))
    println("============================")

    //差集
    val subRDD: RDD[Int] = rdd1.subtract(rdd2)
    println(subRDD.collect().mkString(","))
    println("============================")

    //拉链   (1,3),(2,4),(3,5),(4,6)
    val zipRDD: RDD[(Int, Int)] = rdd1.zip(rdd2)
    val zipRDD1 = rdd1.zip(rdd3)
    println(zipRDD.collect().mkString(","))
    println(zipRDD1.collect().mkString(","))

//交集，并集和差集要求两个数据源数据类型保持一致
//拉链操作两个数据源的类型可以不一致

//Can't zip RDDs with unequal numbers of partitions: List(2, 4)
//两个数据源要求分区数量保持一致
//    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)
//    val rdd2: RDD[Int] = sc.makeRDD(List(3, 4, 5, 6),4)

//Can only zip RDDs with same number of elements in each partition
//两个数据源要求分区中数据数量保持一致
    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6),2)
    val rdd2: RDD[Int] = sc.makeRDD(List(3, 4, 5, 6),2)

4.3.3 Key - Value 类型

1. partitionBy

函数签名 def partitionBy(partitioner: Partitioner): RDD[(K, V)]

函数说明将数据按照指定Partitioner 重新进行分区。Spark 默认的分区器是HashPartitioner

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

val mapRDD: RDD[(Int, Int)] = rdd.map((_, 1))
//RDD => PairRDDFunctions
//隐式转换（二次编译）

//partitionBy根据指定的分区规则对数据进行重分区
mapRDD.partitionBy(new HashPartitioner(2)).saveAsTextFile("output")
//(2,1) (4,1)
//(1,1) (3,1)

####

2. reduceByKey

函数签名 def reduceByKey(func: (V, V) => V): RDD[(K, V)] def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]

函数说明可以将数据按照相同的Key 对Value 进行聚合

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

    //相同的key的数据进行value数据的聚合操作
    //scala语言中一般的聚合操作都是两两聚合，spark基于Scala开发，所以它的聚合也是两两聚合
    // [1,2,3]
    // [3,3]
    // [6]
    //reduceByKey中如果key的数据只有一个，是不会参与运算的
    val reduceRDD: RDD[(String, Int)] = rdd.reduceByKey((x, y) => {
      println(s"x = ${x} , y = ${y}")
      x + y
    })
    //x = 1 , y = 2
    //x = 3 , y = 3
    //(a,6)
    //(b,4)

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 进行分组

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

    //groupByKey: 将数据源中的数据，相同的key的数据分在一个组中，形成一个对偶元组
    //            元组中的第一个元素就是key
    //            元组中的第二个元素就是相同key的value的集合
    val groupRDD: RDD[(String, Iterable[Int])] = rdd.groupByKey()
    groupRDD.collect().foreach(println)

    println("================================")

    val groupRDD1: RDD[(String, Iterable[(String, Int)])] = rdd.groupBy(_._1)
    groupRDD1.collect().foreach(println)

(a,CompactBuffer(1, 2, 3))
(b,CompactBuffer(4))
================================
(a,CompactBuffer((a,1), (a,2), (a,3)))
(b,CompactBuffer((b,4)))

思考一个问题：reduceByKey 和groupByKey 的区别？

从shuffle 的角度：

reduceByKey 和groupByKey 都存在shuffle 的操作，但是reduceByKey 可以在shuffle 前对分区内相同key 的数据进行预聚合（combine）功能，这样会减少落盘的数据量，而groupByKey 只是进行分组，不存在数据量减少的问题，reduceByKey 性能比较高。

从功能的角度：

reduceByKey 其实包含分组和聚合的功能。GroupByKey 只能分组，不能聚合，所以在分组聚合的场合下，推荐使用reduceByKey，如果仅仅是分组而不需要聚合。那么还是只能使用groupByKey

4. aggregateByKey

函数签名 def aggregateByKeyU: ClassTag(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)] 函数说明将数据根据不同的规则进行分区内计算和分区间计算

//将数据根据不同的规则进行分区内计算和分区间计算

val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4),("b", 5), ("a", 6)),2)

    //aggregateByKey存在函数柯里化，有两个函数列表
    //第一个参数列表, 需要传递一个参数，表示为初始值
    //     主要用于当碰见第一个key的时候，和value进行分区内计算
    //第二个参数列表需要传递两个参数：
    //     第一个参数表示分区内计算规则
    //     第二个参数表示分区间计算规则
 rdd.aggregateByKey(0)((x,y) => math.max(x,y), (x,y) => x+y).collect().foreach(println)

 rdd.aggregateByKey(5)((x,y) => math.max(x,y), (x,y) => x+y).collect().foreach(println)

(b,8)	3+5
(a,8)	2+6
(b,10)	5+5
(a,11)	5+6

5. foldByKey

函数签名 def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]

函数说明当分区内计算规则和分区间计算规则相同时，aggregateByKey 就可以简化为foldByKey

val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4),("b", 5), ("a", 6)),2)
    
//rdd.aggregateByKey(0)((x,y) => x+y, (x,y) => x+y).collect().foreach(println)
  rdd.aggregateByKey(0)(_+_, _+_).collect().foreach(println)

    println("===============================================")

    //如果聚合计算时，分区内和分区间计算规则相同，spark提供了简化的方法
    rdd.foldByKey(0)(_+_).collect().foreach(println)

6. combineByKey

函数签名 def combineByKeyC: RDD[(K, C)]

函数说明最通用的对key-value 型rdd 进行聚集操作的聚集函数（aggregation function）。类似于aggregate()，combineByKey()允许用户返回值的类型与输入不一致。

    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4),("b", 5), ("a", 6)),2)
    
    //combineByKey方法需要三个参数
    //第一个参数表示： 将相同的key的第一个数据进行结构转换，实现操作
    //第二个参数表示：分区内计算规则
    //第三个参数表示：分区间计算规则

    //获取相同key的数据的平均值
    //(a,3) (b,4)
    val newRDD: RDD[(String, (Int, Int))] = rdd.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 + t2._2)
      }
    )

    newRDD.collect().foreach(println)

    println("===============================")

    val resultRDD: RDD[(String, Int)] = newRDD.mapValues {
      case (num, cnt) => {
        num / cnt
      }
    }

(b,(12,3))
(a,(9,3))
===============================
(b,4)
(a,3)

思考一个问题：reduceByKey、foldByKey、aggregateByKey、combineByKey 的区别？

reduceByKey: 相同key 的第一个数据不进行任何计算，分区内和分区间计算规则相同 FoldByKey: 相同key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则相同 AggregateByKey：相同key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则可以不相同 CombineByKey:当计算时，发现数据结构不满足要求时，可以让第一个数据转换结构。分区内和分区间计算规则不相同

    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4),("b", 5), ("a", 6)),2)
    /*
    reduceByKey:
           combineByKeyWithClassTag[V](
                  (v: V) => v,    //第一个值不会参与计算
                  func,           //分区内计算规则
                  func,            //分区间计算规则
                  partitioner)

    aggregateByKey:
            combineByKeyWithClassTag[U](
                  (v: V) => cleanedSeqOp(createZero(), v),    //初始值和第一个key的value值进行的分区内数据操作
                  cleanedSeqOp,        //分区内计算规则
                  combOp,               //分区间计算规则
                  partitioner)

    foldByKey:
             combineByKeyWithClassTag[V](
                  (v: V) => cleanedFunc(createZero(), v),   //初始值和第一个key的value值进行的分区内数据操作
                  cleanedFunc,        //分区内计算规则
                  cleanedFunc,        //分区间计算规则
                  partitioner)

    combineByKey:
             combineByKeyWithClassTag(
                  createCombiner,   //相同key烦人第一条数据进行的处理函数
                  mergeValue,       //表示分区内数据的处理函数
                  mergeCombiners,   //表示分区间数据的处理函数
                  defaultPartitioner(self))

     */

    rdd.reduceByKey(_+_)  //  wordcount
    rdd.aggregateByKey(0)(_+_, _+_)   //wordcount
    rdd.foldByKey(0)(_+_)     //wordcount
    rdd.combineByKey(v=>v, (x:Int, y)=>x+y, (x:Int, y:Int)=>x+y)    //wordcunt

7. sortByKey

函数签名
def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]

函数说明在一个(K,V)的 RDD 上调用，K 必须实现 Ordered 接口(特质)，返回一个按照 key 进行排序的

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)

####

8. join

函数签名 def joinW: RDD[(K, (V, W))]

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

    val rdd1: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
    val rdd2: RDD[(String, Int)] = sc.makeRDD(List(("b", 5), ("c", 6), ("a", 4)))
    val rdd3: RDD[(String, Int)] = sc.makeRDD(List(("d", 5), ("c", 6), ("a", 4)))
    val rdd4: RDD[(String, Int)] = sc.makeRDD(List(("a", 5), ("c", 6), ("a", 4)))

    //join: 两个不同数据源的数据，相同的key的value会连接在一起，形成元组
    //      如果两个数据源中key没有匹配上，那么数据不会出现在结果中
    //      如果两个数据源中key有多个相同的，会依次匹配，可能会出现笛卡尔积，数据会几何性增长，会导致性能降低
    val joinRDD: RDD[(String, (Int, Int))] = rdd1.join(rdd2)
    val joinRDD1: RDD[(String, (Int, Int))] = rdd1.join(rdd3)
    val joinRDD2: RDD[(String, (Int, Int))] = rdd1.join(rdd4)

    joinRDD.collect().foreach(println)
    println("===========================")
    joinRDD1.collect().foreach(println)
    println("===========================")
    joinRDD2.collect().foreach(println)

(a,(1,4))
(b,(2,5))
(c,(3,6))
===========================
(a,(1,4))
(c,(3,6))
===========================
(a,(1,5))
(a,(1,4))
(c,(3,6))

9. leftOuterJoin

函数签名 def leftOuterJoinW: RDD[(K, (V, Option[W]))]

函数说明类似于SQL 语句的左外连接

    val rdd1: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
    val rdd2: RDD[(String, Int)] = sc.makeRDD(List(("a", 5), ("b", 6)))

    val leftJoinRDD: RDD[(String, (Int, Option[Int]))] = rdd1.leftOuterJoin(rdd2)
    val rightJoinRDD: RDD[(String, (Option[Int], Int))] = rdd1.rightOuterJoin(rdd2)


    leftJoinRDD.collect().foreach(println)
    println("===========================")
    rightJoinRDD.collect().foreach(println)

(a,(1,Some(5)))
(b,(2,Some(6)))
(c,(3,None))
===========================
(a,(Some(1),5))
(b,(Some(2),6))

10. cogroup

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

   val rdd1: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2), ("c", 3), ("c", 3)))
   val rdd2: RDD[(String, Int)] = sc.makeRDD(List(("a", 5), ("b", 6)))

   val cgRDD: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd1.cogroup(rdd2)

   cgRDD.collect().foreach(println)

(a,(CompactBuffer(1),CompactBuffer(5)))
(b,(CompactBuffer(2),CompactBuffer(6)))
(c,(CompactBuffer(3, 3),CompactBuffer()))

4.4 RDD 行动算子

1. reduce

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

2. collect

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

3. count

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

4. first

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

5. take

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

6. takeOrdered

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

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

    //TODO - 行动算子 reduce
    val result: Int = rdd.reduce(_ + _)
    println(result)

    println("=======================")

    //TODO - 行动算子 collect
    //方法会将不同分区的数据按照分区顺序采集到Driver端内存中，形成数组
    val result1: Array[Int] = rdd.collect()
    println(result1.mkString(","))

    println("=======================")

    //TODO - 行动算子 count
    //统计数据源中数据的个数
    val cnt: Long = rdd.count()
    println(cnt)

    println("=======================")

    //TODO - 行动算子 first
    //获取数据源中数据的第一个
    val first: Int = rdd.first()
    println(first)

    println("=======================")

    //TODO - 行动算子 take
    //获取N个数据
    val take: Array[Int] = rdd.take(3)
    println(take.mkString(","))

    println("=======================")

    //TODO - 行动算子 takeOrdered
    //数据排序后，取N个数据
    val rdd1: RDD[Int] = sc.makeRDD(List(4, 2, 3, 1))
    val takeO: Array[Int] = rdd1.takeOrdered(3)
    println(takeO.mkString(","))

10
=======================
1,2,3,4
=======================
4
=======================
1
=======================
1,2,3
=======================
1,2,3

####

7. aggregate

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

8. fold

函数签名 def fold(zeroValue: T)(op: (T, T) => T): T 函数说明折叠操作，aggregate 的简化版操作

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

    //TODO - 行动算子 aggregate
    val result: Int = rdd.aggregate(0)(_ + _, _ + _)

    //13 + 17
    //aggregateByKey: 初始值只会参与分区内计算
    //aggregate: 初始值会参与分区内计算，并且会参与分区间计算
    // 13 + 17 + 10 = 40
    val result1: Int = rdd.aggregate(10)(_ + _, _ + _)

    println(result)
    println(result1)

    println("=====================")

    //TODO - 行动算子 fold

    val result2: Int = rdd.fold(10)(_ + _)
    println(result2)

10
40
=====================
40

####

9. countByKey

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

    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)
    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)
    val rdd2: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("a", 3)))

    //TODO - 行动算子 countByKey
    val intToLong: collection.Map[Int, Long] = rdd.countByValue()
    //Map(4 -> 1, 2 -> 1, 1 -> 1, 3 -> 1)

    val intToLong1: collection.Map[Int, Long] = rdd1.countByValue()
    //Map(4 -> 1, 2 -> 1, 1 -> 1, 3 -> 1)

    val stringToLong: collection.Map[String, Long] = rdd2.countByKey()
    //Map(a -> 3)

10. save 相关算子

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

// 保存成 Text 文件
rdd.saveAsTextFile("output")

// 序列化成对象保存到文件
rdd.saveAsObjectFile("output1")

// 保存成 Sequencefile 文件
rdd.map((_,1)).saveAsSequenceFile("output2")

11. foreach

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

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

    //TODO - 行动算子 foreach

    //foreach其实是Driver端内存集合的循环遍历方法
    rdd.collect().foreach(println)
    println("======================")
    //foreach其实是Executor端内存数据打印
    rdd.foreach(println)

    //什么是算子： Operator（操作）
    //           RDD方法和Scala集合对象的方法不一样
    //           集合对象的方法都是在同一个节点的内存中完成的
    //           RDD方法可以将计算逻辑发送到Executor端（分布式节点）进行
    //           为了区分不同的处理效果，所以将RDD的方法称之为算子
    //           RDD的方法外部的操作都是在Driver端执行的，而方法内部的逻辑代码是在Executor端执行的
/*
1 2 3 4
======================
2 3 4 1
 */

4.5 RDD 序列化

1. 闭包检查

从计算的角度, 算子以外的代码都是在Driver 端执行, 算子里面的代码都是在Executor 端执行。那么在scala 的函数式编程中，就会导致算子内经常会用到算子外的数据，这样就形成了闭包的效果，如果使用的算子外的数据无法序列化，就意味着无法传值给Executor 端执行，就会发生错误，所以需要在执行任务计算前，检测闭包内的对象是否可以进行序列化，这个操作我们称之为闭包检测。Scala2.12 版本后闭包编译方式发生了改变

2. 序列化方法和属性

从计算的角度, 算子以外的代码都是在Driver 端执行, 算子里面的代码都是在Executor端执行，看如下代码：

object serializable02_function {
  def main(args: Array[String]): Unit = {
    //1.创建 SparkConf 并设置 App 名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
   
    //2.创建 SparkContext，该对象是提交 Spark App 的入口
    val sc: SparkContext = new SparkContext(conf)
    
    //3.创建一个 RDD
    val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello spark", "hive", "mhk"))
    
    //3.1 创建一个 Search 对象
    val search = new Search("hello")
    
    //3.2 函数传递，打印：ERROR Task not serializable
    search.getMatch1(rdd).collect().foreach(println)
    
    //3.3 属性传递，打印：ERROR Task not serializable
    search.getMatch2(rdd).collect().foreach(println)
  
    //4.关闭连接
    sc.stop()
  } 
}

//查询对象
//Scala语法中类的构造参数其实是类的属性，构造参数需要闭包检测，其实就等同于类进行闭包检测
class Search(query:String) extends Serializable {
   def isMatch(s: String): Boolean = {
    s.contains(query)
}
// 函数序列化案例
  def getMatch1 (rdd: RDD[String]): RDD[String] = {
    //rdd.filter(this.isMatch)
    rdd.filter(isMatch)
}
  // 属性序列化案例
  def getMatch2(rdd: RDD[String]): RDD[String] = {
    //rdd.filter(x => x.contains(this.query))
    rdd.filter(x => x.contains(query))
    //val q = query
    //rdd.filter(x => x.contains(q))
  } 
}

3. Kryo 序列化框架

参考地址: https://github.com/EsotericSoftware/kryo

Java 的序列化能够序列化任何的类。但是比较重（字节多），序列化后，对象的提交也比较大。Spark 出于性能的考虑，Spark2.0 开始支持另外一种Kryo 序列化机制。Kryo 速度是Serializable 的10 倍。当RDD 在Shuffle 数据的时候，简单数据类型、数组和字符串类型已经在Spark 内部使用Kryo 来序列化。注意：即使使用Kryo 序列化，也要继承Serializable 接口

object serializable_Kryo {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("SerDemo").setMaster("local[*]")
            // 替换默认的序列化机制
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
            // 注册需要使用 kryo 序列化的自定义类
          .registerKryoClasses(Array(classOf[Searcher]))

    val sc = new SparkContext(conf)
 
    val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello atguigu", "atguigu", "hahah"), 2)

    val searcher = new Searcher("hello")

    val result: RDD[String] = searcher.getMatchedRDD1(rdd)

    result.collect.foreach(println)
  } 	
}

case class Searcher(val query: String) {
  def isMatch(s: String) = {
    s.contains(query)
   }
  def getMatchedRDD1(rdd: RDD[String]) = {
    rdd.filter(isMatch) 
  }
  def getMatchedRDD2(rdd: RDD[String]) = {
    val q = query
    rdd.filter(_.contains(q))
  } 
}

4.6 RDD 依赖关系

1. RDD 血缘关系

RDD 只支持粗粒度转换，即在大量记录上执行的单个操作。将创建RDD 的一系列Lineage（血统）记录下来，以便恢复丢失的分区。RDD 的Lineage 会记录RDD 的元数据信息和转换行为，当该RDD 的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区

    // 1. 读取文件，获取一行一行数据
    //    hello world
    val lines: RDD[String] = sc.textFile("datas/word.txt")

    //打印血缘关系
    println(lines.toDebugString)
    println("===============================")

    // 2. 将一行数据进行拆分，形成一个一个的单词（分词）
    // 扁平化：将整体拆分成个体的操作
    //  "hello world" => hello, world, hello, world
    val words: RDD[String] = lines.flatMap(_.split(" "))

    //打印血缘关系
    println(words.toDebugString)
    println("===============================")

    // 3. 将数据根据单词进行分组，便于统计
    //   (hello, hello, hello) (world, world)
    val wordGroup: RDD[(String, Iterable[String])] = words.groupBy(word => word)

    //打印血缘关系
    println(wordGroup.toDebugString)
    println("===============================")

    // 4. 对分组后的数据进行转换
    //   (hello, hello, hello) (world, world)
    //   (hello, 3)   (world, 2)
    val wordToCount: RDD[(String, Int)] = wordGroup.map {
      case (word, list) => {
        (word, list.size)
      }
    }

    //打印血缘关系
    println(wordToCount.toDebugString)
    println("===============================")

    // 5. 将转换结果采集到控制台打印出来
    val array: Array[(String, Int)] = wordToCount.collect()
    array.foreach(println)


(1) datas/word.txt MapPartitionsRDD[1] at textFile at Spark01_RDD_Dep.scala:16 []
 |  datas/word.txt HadoopRDD[0] at textFile at Spark01_RDD_Dep.scala:16 []
===============================
(1) MapPartitionsRDD[2] at flatMap at Spark01_RDD_Dep.scala:25 []
 |  datas/word.txt MapPartitionsRDD[1] at textFile at Spark01_RDD_Dep.scala:16 []
 |  datas/word.txt HadoopRDD[0] at textFile at Spark01_RDD_Dep.scala:16 []
===============================
(1) ShuffledRDD[4] at groupBy at Spark01_RDD_Dep.scala:33 []
 +-(1) MapPartitionsRDD[3] at groupBy at Spark01_RDD_Dep.scala:33 []
    |  MapPartitionsRDD[2] at flatMap at Spark01_RDD_Dep.scala:25 []
    |  datas/word.txt MapPartitionsRDD[1] at textFile at Spark01_RDD_Dep.scala:16 []
    |  datas/word.txt HadoopRDD[0] at textFile at Spark01_RDD_Dep.scala:16 []
===============================
(1) MapPartitionsRDD[5] at map at Spark01_RDD_Dep.scala:42 []
 |  ShuffledRDD[4] at groupBy at Spark01_RDD_Dep.scala:33 []
 +-(1) MapPartitionsRDD[3] at groupBy at Spark01_RDD_Dep.scala:33 []
    |  MapPartitionsRDD[2] at flatMap at Spark01_RDD_Dep.scala:25 []
    |  datas/word.txt MapPartitionsRDD[1] at textFile at Spark01_RDD_Dep.scala:16 []
    |  datas/word.txt HadoopRDD[0] at textFile at Spark01_RDD_Dep.scala:16 []
===============================
(Spark,1)
(Hello,2)
(Scala,1)

2. RDD 依赖关系

这里所谓的依赖关系，其实就是两个相邻RDD 之间的关系

    // 1. 读取文件，获取一行一行数据
    //    hello world
    val lines: RDD[String] = sc.textFile("datas/word.txt")

    //打印依赖关系
    println(lines.dependencies)
    println("===============================")

    // 2. 将一行数据进行拆分，形成一个一个的单词（分词）
    // 扁平化：将整体拆分成个体的操作
    //  "hello world" => hello, world, hello, world
    val words: RDD[String] = lines.flatMap(_.split(" "))

    //打印依赖关系
    println(words.dependencies)
    println("===============================")

    // 3. 将数据根据单词进行分组，便于统计
    //   (hello, hello, hello) (world, world)
    val wordGroup: RDD[(String, Iterable[String])] = words.groupBy(word => word)

    //打印依赖关系
    println(wordGroup.dependencies)
    println("===============================")

    // 4. 对分组后的数据进行转换
    //   (hello, hello, hello) (world, world)
    //   (hello, 3)   (world, 2)
    val wordToCount: RDD[(String, Int)] = wordGroup.map {
      case (word, list) => {
        (word, list.size)
      }
    }

    //打印依赖关系
    println(wordToCount.dependencies)
    println("===============================")

    // 5. 将转换结果采集到控制台打印出来
    val array: Array[(String, Int)] = wordToCount.collect()
    array.foreach(println)


List(org.apache.spark.OneToOneDependency@434514d8)
===============================
List(org.apache.spark.OneToOneDependency@42b84286)
===============================
List(org.apache.spark.ShuffleDependency@440e3ce6)
===============================
List(org.apache.spark.OneToOneDependency@61bfc9bf)
===============================
(Spark,1)
(Hello,2)
(Scala,1)

相邻的两个RDD的关系称之为依赖关系 val rdd1 = rdd.map( _ * 2 ) 新的RDD依赖于旧的RDD 多个连续的RDD的依赖关系，称之为血缘关系每个RDD会保存血缘关系

3. RDD 窄依赖

窄依赖表示每一个父(上游)RDD 的Partition 最多被子（下游）RDD 的一个Partition 使用，窄依赖我们形象的比喻为独生子女。

class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd)

4. RDD 宽依赖

宽依赖表示同一个父（上游）RDD 的Partition 被多个子（下游）RDD 的Partition 依赖，会引起Shuffle，总结：宽依赖我们形象的比喻为多生。

class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
    @transient private val _rdd: RDD[_ <: Product2[K, V]],
    val partitioner: Partitioner,
    val serializer: Serializer = SparkEnv.get.serializer,
    val keyOrdering: Option[Ordering[K]] = None,
    val aggregator: Option[Aggregator[K, V, C]] = None,
    val mapSideCombine: Boolean = false)
  extends Dependency[Product2[K, V]]

5. RDD 阶段划分

DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，该图形具有方向，不会闭环。例如，DAG 记录了RDD 的转换过程和任务的阶段。 7. RDD 任务划分 RDD 任务切分中间分为：Application、Job、Stage 和Task

Application：初始化一个SparkContext 即生成一个Application；
Job：一个Action 算子就会生成一个Job；
Stage：Stage 等于宽依赖(ShuffleDependency)的个数加1；
Task：一个Stage 阶段中，最后一个RDD 的分区个数就是Task 的个数。

注意：Application->Job->Stage->Task 每一层都是1 对n 的关系。

4.6 RDD 持久化

1. RDD Cache 缓存

RDD 通过Cache 或者Persist 方法将前面的计算结果缓存，默认情况下会把数据以缓存在JVM 的堆内存中。但是并不是这两个方法被调用时立即缓存，而是触发后面的action 算子时，该RDD 将会被缓存在计算节点的内存中，并供后面重用

    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc: SparkContext = new SparkContext(sparkConf)

    val list: RDD[String] = sc.makeRDD(List("hello spark", "hello scala"))

    val flatRDD: RDD[String] = list.flatMap(_.split(" "))

    val mapRDD: RDD[(String, Int)] = flatRDD.map( word => {
          println("**********")
          (word, 1)
      }
    )

    //cache默认的持久化操作，只能将数据保持到内存中，如果想要想要保存到磁盘文件，需要更改存储级别。persist(StorageLevel.DISK_ONLY)
    //mapRDD.cache()

    //持久化操作必须在行动算子执行时完成的
    //连行动都没做哪来的数据缓存呢
    mapRDD.persist(StorageLevel.DISK_ONLY)

    //RDD对象的持久化操作不一定是为了重用
    //在数据执行较长，或数据比较重要的场合也可以采用

    val result: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)

    result.collect().foreach(println)

    println("==================================")

    val groupRDD: RDD[(String, Iterable[Int])] = mapRDD.groupByKey()

    groupRDD.collect().foreach(println)

  }


**********
**********
**********
**********
(scala,1)
(spark,1)
(hello,2)
==================================
(scala,CompactBuffer(1))
(spark,CompactBuffer(1))
(hello,CompactBuffer(1, 1))

存储级别

object StorageLevel {
  val NONE = new StorageLevel(false, false, false, false)  
  val DISK_ONLY = new StorageLevel(true, false, false, false)
  val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2) 
  val MEMORY_ONLY = new StorageLevel(false, true, false, true)
  val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2) 
  val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
  val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2) 
  val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
  val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
  val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
  val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2) 
  val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
}

缓存有可能丢失，或者存储于内存的数据由于内存不足而被删除，RDD 的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于RDD 的一系列转换，丢失的数据会被重算，由于RDD 的各个Partition 是相对独立的，因此只需要计算丢失的部分即可，并不需要重算全部Partition。

Spark 会自动对一些Shuffle 操作的中间数据做持久化操作(比如：reduceByKey)。这样做的目的是为了当一个节点Shuffle 失败了避免重新计算整个输入。但是，在实际使用的时候，如果想重用数据，仍然建议调用persist 或cache。

3. RDD CheckPoint 检查点

所谓的检查点其实就是通过将RDD 中间结果写入磁盘由于血缘依赖过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果检查点之后有节点出现问题，可以从检查点开始重做血缘，减少了开销。对RDD 进行checkpoint 操作并不会马上被执行，必须执行Action 操作才能触发。

    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc: SparkContext = new SparkContext(sparkConf)

    sc.setCheckpointDir("checkpoint")

    val list: RDD[String] = sc.makeRDD(List("hello spark", "hello scala"))

    val flatRDD: RDD[String] = list.flatMap(_.split(" "))

    val mapRDD: RDD[(String, Int)] = flatRDD.map( word => {
          println("**********")
          (word, 1)
      }
    )
    mapRDD.cache()
    mapRDD.checkpoint()

    val result: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)

    result.collect().foreach(println)

    println("==================================")

    val groupRDD: RDD[(String, Iterable[Int])] = mapRDD.groupByKey()

    groupRDD.collect().foreach(println)

  }

cache: 将数据临时存储在内存中进行数据重用

persist：将数据临时存储在磁盘文件中进行数据重用涉及到磁盘IO，性能较低，但是数据安全如果作业执行完毕，临时保存的数据文件就会丢失

checkpoint：将数据长久的保存在磁盘文件中进行数据重用涉及到磁盘IO，性能较低，但是数据安全为了保证数据安全，所以一般情况下会独立执行作业为了能够提高效率，一般情况下，是需要和cache联合使用

3. 缓存和检查点区别

1） Cache 缓存只是将数据保存起来，不切断血缘依赖。Checkpoint 检查点切断血缘依赖。

2） Cache 缓存的数据通常存储在磁盘、内存等地方，可靠性低。Checkpoint 的数据通常存储在HDFS 等容错、高可用的文件系统，可靠性高。

3）建议对checkpoint()的RDD 使用Cache 缓存，这样checkpoint 的job 只需从Cache 缓存中读取数据即可，否则需要再从头计算一次RDD。

4.7 RDD 分区器

Spark 目前支持Hash 分区和Range 分区，和用户自定义分区。Hash 分区为当前的默认分区。分区器直接决定了RDD 中分区的个数、RDD 中每条数据经过Shuffle 后进入哪个分区，进而决定了Reduce 的个数。

只有Key-Value 类型的RDD 才有分区器，非Key-Value 类型的RDD 分区的值是None
每个RDD 的分区ID 范围：0 ~ (numPartitions - 1)，决定这个值是属于那个分区的。

1) Hash 分区：对于给定的key，计算其hashCode,并除以分区个数取余

2)** Range 分区**：将一定范围内的数据映射到一个分区中，尽量保证每个分区数据均匀，而且分区间有序

3) 自定义分区

object Spark01_RDD_Part {
  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc: SparkContext = new SparkContext(sparkConf)

    val rdd: RDD[(String, String)] = sc.makeRDD(List(
      ("RealMadrid", "Kroos"),
      ("Bayren", "Muller"),
      ("RealMadrid", "Benzema"),
      ("ManUnited", "Ronaldo"),
      ("RealMadrid", "Haaland")
    ))

    val partRDD: RDD[(String, String)] = rdd.partitionBy(new MyPartitioner)

    partRDD.saveAsTextFile("output")

    sc.stop()

  }
}

/**
 * 自定义分区器
 * 1. 继承Partitioner
 * 2. 重写方法
 */
class MyPartitioner extends Partitioner{

  //分区数量
  override def numPartitions: Int = 3
  //根据数据的key值返回数据所在的分区索引（从0开始）
  override def getPartition(key: Any): Int = {
    key match {
      case "RealMadrid" => 0
      case "ManUnited" => 1
      case _ => 2
    }
  }
}

4.8 RDD 文件读取与保存

Spark 的数据读取及数据保存可以从两个维度来作区分：文件格式以及文件系统。文件格式分为：text 文件、csv 文件、sequence 文件以及Object 文件；文件系统分为：本地文件系统、HDFS、HBASE 以及数据库

sequence 文件

SequenceFile 文件是Hadoop用来存储二进制形式的key-value 对而设计的一种平面文件(Flat File)。在SparkContext 中，可以调用sequenceFilekeyClass, valueClass。

object 对象文件

对象文件是将对象序列化后保存的文件，采用Java 的序列化机制。可以通过objectFileT: ClassTag函数接收一个路径，读取对象文件，返回对应的RDD，也可以通过调用saveAsObjectFile()实现对对象文件的输出。因为是序列化所以要指定类型。