一、RDD依赖关系
1、RDD的依赖关系分为窄依赖和宽依赖；
2、窄依赖是说父RDD的每一个分区最多被一个子RDD的分区应用，也就是它的出度为1；
3、宽依赖是说父RDD的每一个分区被多个子RDD的分区来应用，也就是它的出度大于等于2；
4、应用在整个过程中，RDD之间形成的产生关系，就叫做血统关系，RDD在没有持久化的时候默认是不保存的，如果需要那么就要根据血统关系来重新计算；
5、应用在执行过程中是分为多个stage来进行的，划分stage的关键就是判断是不是存在宽依赖。从action往前去推整个stage的划分。
二、RDD的持久化
1、RDD的持久化主要通过persist和cache操作来实现。cache操作就相当于storagelevel为MEMORY_ONLY的persist操作。
def persist():this.type=persist(StorageLevel.MEMORY_ONLY)
def cache():this.type=persisit()
2、持久化它的存储等级可以分为：存储的位置（磁盘、内存、非堆内存）、是否序列化、存储的份数（1,2）
三、RDD的检查点机制
1、spark中对于数据的保存除了持久化操作之外，还提供了一种检查点的机制，检查点（本质是通过RDD写入Disk做检查点）是为了通过lineage做容错的辅助，lineage过长会造成容错成本过高，这样就不如在中间阶段做检查点容错，如果之后有节点出现问题而丢失分区，从做检查点的RDD开始重做lineage，就会减少开销。检查点通过将数据写入到HDFS文件系统实现了RDD的检查点功能。
2、cache和checkpoint是由显著区别的，缓存把RDD计算出来然后放在内存中，但是RDD的依赖链（相当于数据库中的redo日志），也不能丢掉，当某个点某个executor宕了，上面cache的RDD就会丢掉，需要通过依赖链重放计算出来，不同的是checkpoint是把RDD保存在HDFS中，是多副本可靠存储，所以依赖链就可以丢掉了，就斩断了依赖链，是通过复制实现的高容错。
3、如果存在以下场景，则比较适合使用检查点机制
a、DAG的lineage过长，如果重算则开销太大（如在PageRank中）。
b、在宽依赖上做checkpoint获得的收益更大。
为当前RDD设置检查点。该函数将会创建一个二进制的文件，并存储到checkpoint目录中，该目录使用SparkContext.setCheckpointDir()设置的。在checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将会全部被移出，对RDD进行checkpoint操作并不会马上被执行，必须执行Action操作才能触发。
4、checkpoint的使用
a、先创建一个RDD
b、设置sparkContext的checkpoint目录，如果是用spark-shell，那么就是sc.setCheckpointDir("...")
c、在RDD上调用checkpoint方法
d、触发RDD的行动操作，让RDD的数据真实写入checkpoint目录。
注意：整个checkpoint的读取是用户透明的
四、键值对RDD数据分区
1、spark支持Hash分区和Range分区，用户也可以自定义分区，Hash分区为当前的默认分区，spark中分区器直接决定了RDD中分区的个数、RDD中每条数据经过shuffle过程属于哪个分区和Reduce的个数，讲一下hash分区和range分区：
hash分区:对于给定的key，计算其hashCode，并除于分区的个数取余--容易造成数据倾斜
range分区:采用的是水塘抽样算法：将数据一部分放在分区中，避免了一个数据倾斜的状态
注意：
a、只有Key-Value类型的RDD才有分区的，非Key-Value类型的RDD分区的值是None；
经过几次计算发现，nocheck每次都不同，check2第一次触发到HDFS上之后，每次都从HDFS上获取，所以都是相同的，而且checkpoint在spark streaming用的较多。
下面我们进行一下分区操作：首先初始化时我们给了8个分区
class CustomerPartitioner(numParts:Int) extends org.apache.spark.Partitioner{
override def numPartitions:Int=numParts
override def getPartition(key:Any):Int={
val ckey:String=key.toString
ckey.substring(ckey.length-1).toInt%numParts
}
}
五、RDD累加器和广播变量
1、累加器
一、累加器的使用
1、首先需要通过sparkcontext去声明一个累加器，方法是accumulator，在声明的过程中需要提供累加器的初始值。
2、你可以在转换操作或者行动上直接使用累加器， 可以通过 += 操作符增加累加器的值，但是不能够读取累加器。
注意： 一般不推荐在转换操作使用累加器。一般推荐在行动操作中去使用。
3、 Driver可以通过累加器.value 操作类读取累加器的值并输出。
二、自定义累加器
1、需要继承AccumulatorV2这个虚拟类，然后提供类型参数
a、增加值类型参数
b、输出值类型参数
// 累加器内部数据结构是否为空
override def isZero: Boolean = ???
// 让SPark框架能够调用copy函数产生一个新的系统的类=累加器实例。
override def copy(): AccumulatorV2[String, util.Set[String]] = ???
// 重置你的累加器数据结构
override def reset(): Unit = ???
// 提供修改累加器的方法
override def add(v: String): Unit = ???
// 用于合并多个分区的累加器实例
override def merge(other: AccumulatorV2[String, util.Set[String]]): Unit = ???
// 通过value方法输出你的累加器最终结果
override def value: util.Set[String] = ???
2、需要创建一个SparkContext
3、需要创建一个自定义累加器实例
4、需要通过SparkContext去注册你的累加器，
val accum = new LogAccumulator
sc.register(accum, "logAccum")
5、需要在转换或者行动操作中使用累加器。
6、在Driver中输出累加器的结果。
2、广播变量
1、广播变量的出现时为了解决只读大对象分发的问题。
2、如果不是广播变量，那么使用的变量会跟分区进行分发，效率比较低。
3、广播变量的使用：
a、通过SparkContext.broadcast(对象) 来声明一个广播变量。
b、通过广播变量的变量名的value方法来获取广播变量的值。
