Spark性能调优(五):缓存与Checkpoint

1.怎么设置RDD缓存？2.RDD缓存有哪些策略？
3.Checkpoint和Cache的区别有哪些？

         Spark程序执行的特性，即延迟执行和基于Lineage最大化的pipeline, 当Spark中由于对某个RDD的action操作触发了作业的时候，会基于Lineage从后往前推，找到该RDD的源头的RDD，然后从前往后计算出结果。很明显，如果对某个RDD执行了多次transformation和action操作，每次action操作触发了作业的时候都会重新从源头RDD处计算一遍来获得该RDD，然后再对这个RDD执行相应的操作。当RDD 本身计算特别复杂和耗时（例如某个计算时长超过半个小时）时，这种方式的性能显然是很差的，此时我们必须考虑对计算结果的数据进行持久化。
        持久化也就是将计算出来的RDD，根据配置的持久化级别，保存到内存或磁盘中，以后每次对该RDD进行算子操作时，都会直接从内存或磁盘中提取持久化的RDD数据然后执行算子，而不会从源头处重新计算一遍这个RDD，再执行算子操作，这无疑会提高程序执行的效率。
        我们建议对多次使用的RDD，计算特别复杂和耗时，或计算链条特别长的RDD，进行持久化。我们只需要对希望缓存的RDD persist或cache方法进行标记，这两种方法的区别在于，cache是将RDD 持久化到内存里，而persist可以指定不同的持久化级别。在标记之后，当由Action触发job导致了该RDD的计算后，计算结果就会根据指定的持久化级别被持久化到内存或磁盘中。
        Spark提供了不同的持久化级别，以满足内存使用率和CPU效率的均衡。如何选择合适的持久化级别呢？
我们遵循的准则通常是，优先考虑内存，内存放不下就考虑序列化后放在内存里，尽量不要存储到磁盘上，因为一般RDD的重新计算比从磁盘中读取更快，只有在需要更快的恢复时才使用备份级别（所有的存储级别都通过可以重新计算提供了全面的容错性，但是备份级别允许我们继续在RDD的备份上执行任务而无须重新计算丢失的分区）。
按照顺序来讲，我们通常按照以下方式来选择：
        采用默认情况下性能最高的MEMORY_ONLY。该持久化级别下，不需要进行序列化与反序列化操作，就避免了这部分的性能开销；对这个RDD的后续算子操作，都是基于纯内存中的数据的操作，不需要从磁盘文件中读取数据，性能也很高；而且不需要复制一份数据副本，并远程传送到其他节点上。但是这里必须要注意的是，使用该持久化级别的前提是，集群的内存必须足够大，可以绰绰有余地存放下整个RDD的所有数据，如果RDD中数据比较多时（比如几十亿条），直接用这种持久化级别，可能会导致JVM的OOM内存溢出异常。
        如果使用MEMORY_ONLY级别时发生了内存溢出，那么建议尝试使用MEMORY_ONLY_SER级别。该级别会将RDD数据序列化后再保存在内存中，此时每个partition仅仅是一个字节数组而已，大大减少了对象数量，并降低了内存占用。这种级别比MEMORY_ONLY多出来的性能开销，主要就是序列化与反序列化的开销。但是后续算子可以基于纯内存进行操作，因此性能总体还是比较高的。此外，可能发生的问题同上，如果RDD中的数据量过多的话，还是可能会导致OOM内存溢出的异常。
        如果纯内存的级别都无法使用，那么建议使用MEMORY_AND_DISK_SER策略，而不是MEMORY_AND_DISK策略。因为此时RDD的数据量很大，内存无法完全放下，需要通过序列化来节省内存和磁盘的空间开销。该策略会优先尽量尝试将数据缓存在内存中，内存缓存不下才会写入磁盘。
通常不建议使用DISK_ONLY级别。因为完全基于磁盘文件进行数据的读写，会导致性能急剧降低，有时还不如重新计算一次该RDD。
        通常不建议使用后缀为_2的备份级别。因为该级别必须将所有的数据都复制一份副本，并发送到其他节点上，而数据复制以及网络传输会导致较大的性能开销，除非是作业的高可用性要求很高，否则不建议使用。
这里需要重点指出的是，我们推荐尝试使用OFF_HEAP的方式，因为该方式将RDD数据持久化到了Alluxio中（曾用名Tachyon），而不是executor的内存中。OFF_HEAP方式有以下优势：允许多个executors共享同一个内存池；显著地减少了垃圾回收的开销; 如果某个executors崩溃，缓存的数据不会丢失。事实上，Alluxio带来的不止这些，它是一套基于内存的分布式文件存储系统，为跨程序跨框架地共享内存内数据提供了一套方案。更详细的情况请查阅Alluxio官网。
        需要说明的是，持久化会最大化的保留整个RDD的所有分区的数据，但是如果当前的计算需要内存空间而空闲内存不够的话，那么持久化在内存中的数据必须让出空间，此时如果RDD的持久化级别指定了可以把数据放在磁盘上，那么部分分区数据就可以从内存转入磁盘，否则的话数据就会丢失；Spark的持久化也具有容错性，当持久化的RDD的某一分区丢失后，后续计算该分区后也会自动重新计算并持久化该分区；同时，Spark也有自己的一套机制，自动地监控每个节点上的缓存使用率，并通过LRU算法删除过时的缓存的数据（当然我们也可以手动使用RDD.unpersist()方法来删除）。也就是说，持久化的数据在执行时不一定被有效地持久化了，要想知道欲持久化的RDD有没有被正确地持久化，可以通过查看WebUi的Executors页面
        把数据通过persist或cache持久化到内存或磁盘中，虽然是快速的但不是最可靠的，Checkpoint机制的产生就是为了更加可靠地持久化数据以复用 RDD 计算数据，我们常常针对整个 RDD 计算链条中特别需要数据持久化的环节启用Checkpoint机制来确容错和高可用性。
        我们可以通过调用SparkContext.setCheckpointDir方法来指定Checkpoint时持久化的RDD的数据的存放位置。作为最佳实践，在Checkpoint 中可以指定把数据以多副本的方式存放在本地或HDFS中，（在生产环境下通常是放在HDFS，从而天然地借助 HDFS 高容错和高可靠的特征完成最大化的可靠的持久化）；同时为了提高效率，可以指定多个目录；
        需要说明的是，checkpoint同persist一样是惰性执行的，在对某个RDD标记了需要Checkpoint后，并不会立即执行，只有在后续有Action触发job从而导致了该RDD的计算，且在这个Job执行完成后，才会从后往前回溯找到标记了Checkpoint的RDD，然后新启动一个Job来执行具体的Checkpoint, 所以一般都会对需进行checkpoint的RDD先进行persist标记，从而把该RDD的计算结果持久化到内存或者磁盘上，以备checkpoint复用。


欢迎关注笔者微信公众号“三角兽”，了解更多数学、算法、大数据干货文章。

路过学习一下

补充内容 (2021-3-31 18:15):
優惠券印刷