2.1.7、Cache&Checkpoint&Broadcast&Accumulate

2.1.7.1、Cache

2.1.7.1.1、Cache原理

RDD是通过iterator进行计算的。当然该方法是内部调用的，不会暴露给用户使用；

1、CacheManager通过BlockManager从Local或者Remote获取数据，然后通过RDD的compute进行计算（这个时候需要考虑是否进行checkpoint），根据数据的读取策略，BlockManager首先会从本地获取，如果读取不到，则从远程获取。

2、在计算的时候，会先看当前的RDD是否进行了Checkpoint,如果进行了CK，就会直接读取，否则进行计算。另外因为RDD需要进行缓存，所以计算结果就会通过BlockManager再次进行持久化。

3、根据持久化策略，要么存储到磁盘，要么存储到内存；如果只是缓存到磁盘中国，就直接使用BlockManager.doPut方法写入（这个时候需要考虑Replication）;如果是写入内存，那么会使用MemoryStore.unrollSafely尝试安全将数据保存到内存中，如果内存不够，就会使用一个方法来整理一部分空间。

4、当直接通过RDD的compute进行计算的时候，可能需要考虑checkpoint。

2.1.7.2、Checkpoint

2.1.7.2.0、什么时候需要Checkpoint

1、当计算中有大量的RDD，且本身计算特别复杂且耗时，这个时候需要考虑将计算结果进行持久化

2、需要对曾经计算的结果数据进行复用的时候，可以提升效率

3、当采用persist把数据放到内存或者磁盘时，可能会造成数据丢失

针对以上3种场景，引入了checkpoint来更加可靠持久化数据，可以指定数据放到本地且多副本的方式（生产一般是存储于HDFS）;另一方面确保了RDD复用计算的可靠性，最大程度保证数据安全；通过对RDD启动Checkpoint机制来实现容错和高可用。

2.1.7.2.1、Checkpoint原理

RDD在进行计算之前会先看一下是否有checkpoint，如果有则不需要计算，如果没有被持久化，则进行compute

1、通过Sparkcontext.setCheckpointDir方法设置Checkpoint路径。（这里会进行检查，如果程序在集群中运行，但是目录设置的是本地，那么会记录一个警告，然后driver会试图从它的本地文件系统重建RDD的checkpoint检测点，因为checkpoint文件检查点不正确。实际上应该是在Executor机器上，因为计算是发生在Executor端）

2、如果某个RDD需要进行checkpoint，那么就会生成RDDCheckpointData对象，首先会调用doCheckpoint方法，其内部会通过checkpoint方法会标记RDD的检查点，移除所有引用的父RDDS，来截断RDD血统，并保存到对应的目录中。

当RDD上运行过一个Job后，就会触发RDDCheckpointData.doCheckpoint方法，把路径信息广播给所有的Executor,然后其内部会调用runJob来执行当前的RDD的数据写到checkpoint目录中，同时会产生ReliableCheckpointRDD实例，并调用writeRDDToCheckpointDirectory方法进行checkpoint工作

3、在进行RDD的checkpoint时候，它所依赖的所有的RDD都会从计算链条上清空掉。

4、由于checkpoint是懒加载的，必须得有Job的执行，当Job执行完成之后，才会从后往前回溯哪个RDD进行了checkpoint标记，然后对标记过的RDD再启动一个Job执行具体的checkpoint过程。

2.1.7.2.2、Checkpoint Vs Persist Vs Cache

Cache:底层其实调用的就是Persist(StorageLevel.MEMORY_ONLY),即把数据缓存到内存中

Persist：根据配置的StorageLevel存储策略，将数据缓存到内存或者磁盘。

Checkpoint:通常是将数据持久化到HDFS上,主要用于故障恢复容错保障，且根据上面对checkpoint的原理介绍，checkpoint会切断RDD之间的血缘关系，而persist和cache则会保留RDD之间的血缘关系。从另一方面讲Checkpoint保存数据的文件是不会随着应用程序的结束而删除，除非用户手动删除，而persist或cache则会在程序结束后自动删除。

2.1.7.3、Broadcast

2.1.7.3.1、场景

通常是将数据从一个节点上发送到其他节点上，且存储在内存或者磁盘上，所以不适合存放太大的数据，否则会造成网络I/O压力过大。一般用于处理配置文件、通用的DataSet、数据结构等。

Spark2.X版本中目前存在的Broadcast类型仅有TorrentBroadcast(可能会有网络压力)，1.6版本还有HttpBroadcast（见SPARK-12588），但其可能会有单点压力。

2.1.7.3.2、原理

首先其在SparkContext初始化的时候会创建SparkEnv,这个时候开始初始化广播相关的类。即入口为：

1、其首先会创建BroadcastManager，内部会初始化TorrentBroadcastFactory，该工厂类用于实例化TorrentBroadcast。

2、当调用sc.broadcast进行广播RDD或者其他数据时，那么底层会调用TorrentBroadcast，并首先将其数据写入到Driver端，底层是通过BlockManager进行操作，并将数据会被分成4MB大小的块，然后将块信息上报到BlockManagerMaster中

3、当Executor端的task需要使用到广播变量值的时候，那么会调用value方法获取。那么底层其实是会先进行判断，如果该Executor是首次获取该值的话，那么会向Driver进行请求，当拿到结果后会写入本地，也会像Driver端操作一样，通过BlockManager进行管理，将数据分块上报到全局的BlockManagerMaster中去；如果不是第一次获取且本地存在，则直接读取本地，否则根据随机洗牌向其他的DataServer(Executor/Driver)获取。

管理及生命周期：底层通过BlockManager管理，由于Spark应用程序可能会存在多个Job，并不是每个Job都需要广播变量，也可能一个Job需要很多广播变量，那么其具体生命周期可以手动指定来销毁。

注意：Broadcast变量是只读的，其主要是为了保障数据一致性。避免了容错、更新顺序逻辑的考虑

2.1.7.4、Accumulate

1、分布式全局只写的数据结构、用于数据的累加。只能在Driver端进行读取

2、累加器更新只在action动作内执行，spark保证每个任务对累加器的更新只应用一次，即重新启动的任务不会更新该值；在transformations转换中，如果重新执行任务或者Stage，那么每个任务的更新可能会执行多次。