在Spark应用实行过程中，某些需要多次利用或者重新计算的数据会举行缓存，为后续更多的计算利用复用，避免了再次计算，从而淘汰应用的实行时间，加快整体计算进度。
那么，在计算过程中，哪些数据是需要被缓存的。接下来我们通过一个简单的例子来答复一下。
如下示例，首先对输入数据举行map()计算，得到 mappedRDD，然后再对 mappedRDD 分别举行两种计算：reduceByKey + foreach(println) 和 groupByKey + foreach(println)。
- //输入数据
- var inputRDD = sc.parallelize(Array[(Int,String)])(
- （1,"a"）,（2,"b"）,（3,"c"）,（4,"d"）,（5,"e"）,（3,"f"）,（2,"g"）,（1,"h"）,（2,"i"）,
- ),3)
- val mappedRDD = inputRDD.map(r => (r._1 + 1, r._2))
- // mappedRDD.cache()
- val reducedRDD = mappedRDD.reduceByKey((x,y) => x + "_" + y, 2)
- reducedRDD.foreach(println)
- val groupedRDD = mappedRDD.groupByKey(3).mapValues(V => V.toList)
- groupedRDD.foreach(println)
不丢脸出，在这两个job中，inputRDD => mappedRDD 的计算流程都是一样的，理论上第二个job可以直接从 mappedRDD 开始举行计算的。
我们在代码中取掉mappedRDD.cache()语句的解释，声明一下mappedRDD需要被缓存。
需要注意的是：cache() 利用表现将数据直接写入到内存中，而且cache() 是lazy利用，并不会立即实行的，只有在第一个job运行时才将要缓存的数据写入内存中。
- //输入数据
- var inputRDD = sc.parallelize(Array[(Int,String)])(
- （1,"a"）,（2,"b"）,（3,"c"）,（4,"d"）,（5,"e"）,（3,"f"）,（2,"g"）,（1,"h"）,（2,"i"）,
- ),3)
- val mappedRDD = inputRDD.map(r => (r._1 + 1, r._2))
- mappedRDD.cache()
- val reducedRDD = mappedRDD.reduceByKey((x,y) => x + "_" + y, 2)
- reducedRDD.foreach(println)
- val groupedRDD = mappedRDD.groupByKey(3).mapValues(V => V.toList)
- groupedRDD.foreach(println)
但是带来的代价就是会只占用很多内存空间来缓存数据，试想，假如mappedRDD 包罗了上亿个record，那么其存储将带来很大的内存消耗。这时设置缓存并非最优解，需要权衡计算成本和存储成本。
在本例中，map()利用的计算逻辑很简单，只需要少量的计算成本，假如mappedRDD需要很大的存储空间时，那么就不会对其举行缓存。
缓存机制实际上是一种用空间来换时间的方式，那么如何判断数据是否应该被设置缓存呢？
在 spark 缓存机制中，只要满足如下三个条即可：
（1）会被重复利用的数据。而且被重复利用次数越多，那么缓存带来的性价比也会越高。一样寻常来说，迭代型和交互型的应用会比较适合。
（2）数据不适合过大。因为数据量级过大，会占用大量的存储空间，导致内存不足，会直接降低数据计算可利用的空间（可参考上一篇spark内存的介绍）。固然说缓存过大也可以存放在磁盘中，但是磁盘IO的成本也比较高，甚至不如重新计算的成本低。
（3）非重复缓存的数据。什么意思呢？其实就是假如缓存了某RDD，那么与其存在一对一血缘的父RDD就不需要缓存了。在本例中，假如对 mappedRDD 举行了缓存，那么就没有必要对 inputRDD 举行缓存了。除非有新的job会重复用到inputRDD且不会用到 mappedRDD。
除此之外，不但RDD可以被缓存，广播数据和task计算结果数据也可以被缓存。
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。 |
