什么是RDD—> 弹性分布式数据集RDD

RDD（Resilient Distributed Dataset）叫做分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中，后续的查询能够重用工作集，这极大地提升了查询速度。

RDD五大特性

1. 一组分片（Partition），即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。

1. 一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的，每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合，不需要保存每次计算的结果。

1. RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。

1. 一个Partitioner，即RDD的分片函数。当前Spark中实现了两种类型的分片函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量，也决定了parent RDD Shuffle输出时的分片数量。

1. 一个列表，存储存取每个Partition的优先位置（preferred location）。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念，Spark在进行任务调度的时候，会尽可能地将计算任务分配到其所要处理数据块的存储位置。

创建RDD

1、由一个已经存在的Scala集合创建。

 val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8))

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

val rdd2 = sc.textFile("hdfs://node1.sfy.cn:9000/words.txt")

(List(1,2,3,4,5,6),2) 2指两个分区

RDD编程API

Transformation 算子操作 只是记录了算子的操作（没有记录数据操作）

RDD中的所有转换都是延迟加载的，也就是说，它们并不会直接计算结果。相反的，它们只是记住这些应用到基础数据集（例如一个文件）上的转换动作。只有当发生一个要求返回结果给Driver的动作时，这些转换才会真正运行。这种设计让Spark更加有效率地运行。

常用的Transformation：

Action 触发读数据，计算数据

WordCount中的RDD

WordCount过程一共会产生8个RDD
1 . sc.textFile(“F://words.txt”)


查看rdd数量 rdd.toDebugString

2 . sc.textFile(“F://words.txt”).flatMap(.split(“ “))

3 . sc.textFile(“F://words.txt”).flatMap(.split(“ “)).map((,1))


4 . sc.textFile(“F://words.txt”).flatMap(.split(“ “)).map((,1)).reduceByKey(+)
.sortBy(._2,false)

5.sc.textFile(“F://words.txt”).flatMap(.split(“ “)).map((,1)).reduceByKey(+)
.sortBy(_._2,false).saveAsTextFile(“hdfs://cdh1:9000/out0001”)

Cache操作

cache()方法表示：使用非序列化的方式将RDD的数据全部尝试持久化到内存中，它可以将前面的计算结果缓存，但并不是立即缓存，而是在触发action操作后进行，cache()只是一个transformtion，是lazy的，必须通过一个action触发，才能真正的将该RDD cache到内存中。
创建RDD

action算子操作

创建RDD并存入cache中

再次调用action算子操作，比较两次执行的时间长短，发现存入cache后，其调用速度明显提高

释放

Checkpoint执行及优化

算子较多时，可能出现错误，需要设置一个还原点，还原点会清除掉以前的血统，
作用：恢复之前的计算结果，同时清除RDD之间的依赖关系
1.设置文件将来用来写入计算后的RDD
sc.setCheckDir(“hdfs://cdh1:9000/ck2019114”)

2.读取文件
val rdd = sc.textFile(“hdfs://cdh1:9000/test”)
在此执行cache()可减少额外的工作量

3.对rdd设置checkpoint
rdd.checkpoint
checkpoint是一个transformation

4.查看checkpoint后的结果
rdd.collect

发现触发了两次计算：
1.读取hdfs上的test的文件内容
2.spark要往hdfs上写刚计算(使用了算子就称为计算)后的结果，这种任务是额外的工作量。

checkpoint优化

怎么解决这个问题？（消除任务2）

解决方案：先把rdd进行缓存，rdd.cache() ，再进行checkPoint，再进行collect 就不会触发第二次计算了。
（先进行缓存再进行checkpoint）

RDD的依赖关系

RDD和它依赖的父RDD（s）的关系有两种不同的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。

窄依赖

窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用
总结：窄依赖我们形象的比喻为独生子女

宽依赖

宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition
总结：窄依赖我们形象的比喻为超生

Lineage

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

RDD的缓存

Spark速度非常快的原因之一，就是在不同操作中可以在内存中持久化或缓存个数据集。当持久化某个RDD后，每一个节点都将把计算的分片结果保存在内存中，并在对此RDD或衍生出的RDD进行的其他动作中重用。这使得后续的动作变得更加迅速。RDD相关的持久化和缓存，是Spark最重要的特征之一。可以说，缓存是Spark构建迭代式算法和快速交互式查询的关键。

RDD缓存方式

RDD通过persist方法或cache方法可以将前面的计算结果缓存，但是并不是这两个方法被调用时立即缓存，而是触发后面的action时，该RDD将会被缓存在计算节点的内存中，并供后面重用。

通过查看源码发现cache最终也是调用了persist方法，默认的存储级别都是仅在内存存储一份，Spark的存储级别还有好多种，存储级别在object StorageLevel中定义的。

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

DAG的生成

DAG(Directed Acyclic Graph)叫做有向无环图，原始的RDD通过一系列的转换就就形成了DAG，根据RDD之间的依赖关系的不同将DAG划分成不同的Stage，对于窄依赖，partition的转换处理在Stage中完成计算。对于宽依赖，由于有Shuffle的存在，只能在parent RDD处理完成后，才能开始接下来的计算，因此宽依赖是划分Stage的依据。
Join有可能是宽依赖，也有可能是窄依赖，如果在join前已经进行过分组的那么就是窄依赖。
宽依赖：执行过程中有shuffle的都是宽依赖，宽依赖是划分Stage的依据
窄依赖：一对一
一个shuffle对应一个Stage，一个Stage两个阶段，两个Stage三个阶段

Spark RDD API

map是对每个元素操作, mapPartitions是对其中的每个partition操作

1、**mapPartitionsWithIndex : 把每个partition中的分区号和对应的值拿出来, 看源码

**

val func = (index: Int, iter: Iterator[(Int)]) => {
  iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
}
val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
rdd1.mapPartitionsWithIndex(func).collect

2、aggregate

聚合 zeroValue：初始值
seqOp：内部聚合
combOp：外部聚合

def func1(index: Int, iter: Iterator[(Int)]) : Iterator[String] = {
  iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
}
val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
rdd1.mapPartitionsWithIndex(func1).collect

注意：aggregate是action操作, 第一个参数是初始值, 二:是2个函数[每个函数都是2个参数(第一个参数:先对个个分区进行合并, 第二个:对个个分区合并后的结果再进行合并), 输出一个参数]
###0 + (0+1+2+3+4 + 0+5+6+7+8+9)

rdd1.aggregate(0)(_+_, _+_)
rdd1.aggregate(0)(math.max(_, _), _ + _)
###5和1比, 得5再和234比得5 --> 5和6789比,得9 --> 5 + (5+9)
rdd1.aggregate(5)(math.max(_, _), _ + _)
val rdd2 = sc.parallelize(List("a","b","c","d","e","f"),2)
def func2(index: Int, iter: Iterator[(String)]) : Iterator[String] = {
  iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
}
rdd2.aggregate("")(_ + _, _ + _)
rdd2.aggregate("=")(_ + _, _ + _)
val rdd3 = sc.parallelize(List("12","23","345","4567"),2)
rdd3.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y)
val rdd4 = sc.parallelize(List("12","23","345",""),2)
rdd4.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)
val rdd5 = sc.parallelize(List("12","23","","345"),2)
rdd5.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)

初始值和分区中的第一个元素相比，拿到结果后，使其和当前分区中另一个元素比，得到最小值

第一轮比较 拿到结果后和分区中另一元素比较
math.min(“”.length, “12”.length ) 的结果是：0 , math.min(“”.length, “12”.length ).toString=”0”
第二轮比较（与分区中第二个元素相比）
math.min(“0”.length, “23”.length ) 的结果是1 math.min(“0”.length, “23”.length ).toString=”1”
初始值和第一个分区中元素相比，最终结果为 “1”
与第二分区元素相比
math.min(“”.length, “345”.length) 的结果是：0 , math.min(“0”.length, “”.length) 的结果是：0
最后聚合 “1”+”0” = 10

math.min(“”.length, “12”.length ) 的结果是：0 , math.min(“0”.length, “23”.length ) 的结果是：1
math.min(“”.length, “”.length) 的结果是：0 , math.min(“0”.length, “345”.length) 的结果是：1
最后聚合 “1”+”1” = 11

注意： 0.toString.length=1

0.toString=”0” “”.length=0

3、aggregateByKey

val pairRDD = sc.parallelize(List( ("cat",2), ("cat", 5), ("mouse", 4),("cat", 12), ("dog", 12), ("mouse", 2)), 2)
def func2(index: Int, iter: Iterator[(String, Int)]) : Iterator[String] = {
  iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
}
pairRDD.mapPartitionsWithIndex(func2).collect
pairRDD.aggregateByKey(0)(math.max(_, _), _ + _).collect
pairRDD.aggregateByKey(100)(math.max(_, _), _ + _).collect

4、checkpoint

sc.setCheckpointDir("hdfs://MIN1:9000/ck")
val rdd = sc.textFile("hdfs://MIN1:9000/wc").flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)
rdd.checkpoint
rdd.isCheckpointed
rdd.count
rdd.isCheckpointed
rdd.getCheckpointFile

5、coalesce,[koules] 相当于repartition 重新分区

coalesce：重置分区；当分区数量减少的时候，可以将参数shuffle设置为false,与上面的不同之处在于有shuffle参数
//如果重分区的数目大于原来的分区数，那么必须指定shuffle参数为true

val rdd1 = sc.parallelize(1 to 10, 10)
val rdd2 = rdd1.coalesce(2, false)
rdd2.partitions.length

6、collectAsMap : Map(b -> 2, a -> 1)

val rdd = sc.parallelize(List(("a", 1), ("b", 2)))
rdd.collectAsMap

7、combineByKey : 和reduceByKey是相同的效果

第一个参数x:原封不动取出来, 第二个参数:是函数, 局部运算, 第三个:是函数, 对局部运算后的结果再做运算

###每个分区中每个key中value中的第一个值, (hello,1)(hello,1)(good,1)—>(hello(1,1),good(1))—>x就相当于hello的第一个1, good中的1

val rdd1 = sc.textFile("hdfs://master:9000/wordcount/input/").flatMap(_.split(" ")).map((_, 1))
val rdd2 = rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
rdd1.collect
rdd2.collect

当input下有3个文件时(有3个block块, 不是有3个文件就有3个block, ), 每个会多加3个10

val rdd3 = rdd1.combineByKey(x => x + 10, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
rdd3.collect

val rdd4 = sc.parallelize(List("dog","cat","gnu","salmon","rabbit","turkey","wolf","bear","bee"), 3)
val rdd5 = sc.parallelize(List(1,1,2,2,2,1,2,2,2), 3)
val rdd6 = rdd5.zip(rdd4)
val rdd7 = rdd6.combineByKey(List(_), (x: List[String], y: String) => x :+ y, (m: List[String], n: List[String]) => m ++ n)

8、countByKey

val rdd1 = sc.parallelize(List(("a", 1), ("b", 2), ("b", 2), ("c", 2), ("c", 1)))
rdd1.countByKey
rdd1.countByValue

9、filterByRange

val rdd1 = sc.parallelize(List(("e", 5), ("c", 3), ("d", 4), ("c", 2), ("a", 1),("b", 6)))
val rdd2 = rdd1.filterByRange("b", "d")
rdd2.collect

10、flatMapValues : Array((a,1), (a,2), (b,3), (b,4))

val rdd3 = sc.parallelize(List(("a", "1 2"), ("b", "3 4")))
val rdd4 = rdd3.flatMapValues(_.split(" "))
rdd4.collect

11、foldByKey

val rdd1 = sc.parallelize(List("dog", "wolf", "cat", "bear"), 2)
val rdd2 = rdd1.map(x => (x.length, x))
val rdd3 = rdd2.foldByKey("")(_+_)

val rdd = sc.textFile("hdfs://MIN1:9000/wc").flatMap(_.split(" ")).map((_, 1))
rdd.foldByKey(0)(_+_)

12、foreachPartition

val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)
rdd1.foreachPartition(x => println(x.reduce(_ + _)))

Scala中foreach和map方法的比较：

两者都是用于遍历集合对象，并对每一项执行的指定方法。两者的差异在于map有返回值，Foreach单纯的遍历，无返回值

13、keyBy : 以传入的参数做key

val rdd1 = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"), 3)
val rdd2 = rdd1.keyBy(_.length)   拿出每个单词的长度作为Key
rdd2.collect

val rdd2 = rdd1.keyBy(_(0)).collect 拿出每个单词的首字母作为Key

14、keys values

val rdd1 = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", "eagle"), 2)
val rdd2 = rdd1.map(x => (x.length, x))   (单词长度，单词)
rdd2.keys.collect
rdd2.values.collect

rdd2.keys.collect

rdd2.values.collect