Spark设计原理(二)——数据缓存机制

左手的ㄟ右手 2024-04-06 11:51 25阅读 0赞

## 一.复杂应用的缓存执行计划 ##

val 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))
    
    val reducedByKeyRDD = mappedRDD.reduceByKey((x, y) => x + "_" + y,2)
    val groupedByKeyRDD = mappedRDD.groupByKey().mapValues(v => v.toList)
    
    reducedByKeyRDD.foreach(println)
    groupedByKeyRDD.foreach(println)
    
    val joinedRDD = reducedByKeyRDD.join(groupedByKeyRDD)
    
    joinedRDD.foreach(println)

**图片截图出自 18080端口的History Server**

[（三）5：spark UI详解\_猿来如此dj的博客-CSDN博客\_spark ui][5_spark UI_dj_-CSDN_spark ui]

![06ee7659886146019e635fa3ddd6a534.png][]

job0：  
![64942369bb114aab805cfa4dfb61c4ec.png][]  
job1：  
![903fb1f445a642b1950c73a3756cd813.png][]  
job2：  
![73603bf32015414798607a968ad04972.png][]  
没使用缓存也会有skipped的原因：Spark task和Stage的跳过执行[Spark调度系列-----5.Spark task和Stage的跳过执行(ui显示task和stage skipped)\_heayin123的博客-CSDN博客][Spark_-----5.Spark task_Stage_ui_task_stage skipped_heayin123_-CSDN]

## 使用缓存 ##

val mappedRDD = inputRDD.map(r => (r._1 + 1, r._2))
    
    mappedRDD.cache()
    
    val reducedByKeyRDD = mappedRDD.reduceByKey((x, y) => x + "_" + y,2)
    val groupedByKeyRDD = mappedRDD.groupByKey().mapValues(v => v.toList)
    
    reducedByKeyRDD.cache()
    groupedByKeyRDD.cache()
    
    reducedByKeyRDD.foreach(println)
    groupedByKeyRDD.foreach(println)
    
    val joinedRDD = reducedByKeyRDD.join(groupedByKeyRDD)
    
    joinedRDD.foreach(println)

![4724dd85819641578b2b98e7fd17d4e5.png][]

job0：  
![fb59762062374a38984540c05613cd47.png][]

job1：  
![bd531600b5fc487abc7ef33bcc06539c.png][]

job2：  
![6a1a00367c294ae6a7cfa2d50a65ec5f.png][]

## 二.缓存数据的写入和读取方法 ##

写入方法

上方例子中第一个缓存：  
![32d9a6e26a5040249576ce58db06a4ed.png][]

val mappedRDD = inputRDD.map(r => (r._1 + 1, r._2))
    mappedRDD.cache()
    val reducedByKeyRDD = mappedRDD.reduceByKey((x, y) => x + "_" + y,2)

map()，cache()，shuffle前的combine()的执行顺序是怎么样的呢？  
如果map()操作后，先combine()的话，则mappedRDD的数据会丢失，不会缓存。所以是先缓存，再combine()。  
所以原数据有9条记录，而shuffle write有8条记录，是因为先缓存了，再combine()，combine()后少了一条记录。所以缓存的数据大小可能比shuffle write的大小大。（并且shuffle会进行序列化，也会减少大小）

缓存的数据872B， shuffle write的数据301B：

![6e7952d9409d4d65b8b404a6f2fe1a71.png][]

写入的细节：  
每个Executor都有个区域进行缓存，由blockManager管理。  
将分区数据写入到memoryStore中，memoryStore包含LinkedHashMap，由双向链表实现。  
key为blockId（rddId+partitionId），value为分区数据。

上方例子中第二个缓存：  
![85c2bfaa9d514bbeb243c568a38c45be.png][]

读取方法  
首先会寻找本地的blockManager去寻找缓存数据，如果在另一个结点，需通过远程访问。

##    三.缓存数据的替换和回收 ##

自动缓存替换  
**spark采用了LRU算法（替换当前最久未被使用的RDD）**

用LinkedHashMap来实现LRU，由双向链表实现。  
最近插入或者读取的分区数据放在表头，尾部的数据就是当前最久未被使用的，替换时直接删掉尾部就行。

用户主动回收  
使用 unpersist()，这个操作是立即执行的。 放在job执行之后。

##    四.spark缓存与mapreduce缓存对比 ##

mapreduce的DistributedCache用于缓存job运行所需要的文件，不是用来存储中间结果的。

而且DistributedCache是将缓存文件存在本地磁盘上，不是内存中。

spark缓存的缺点：  
**1.缓存的RDD不能修改。  
2.spark难以获取缓存rdd的生命周期，难以精确的缓存替换  
3.application之间不能共享缓存，可以用Alluxio来解决。**

[5_spark UI_dj_-CSDN_spark ui]: https://blog.csdn.net/weixin_43930865/article/details/108704741
[06ee7659886146019e635fa3ddd6a534.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/ed94fb4f601d4cb68b44de427baf183c.png
[64942369bb114aab805cfa4dfb61c4ec.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/c52d41f017f04f17869cc642addf41da.png
[903fb1f445a642b1950c73a3756cd813.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/b17e971e726a42c1991552b65f2e7d22.png
[73603bf32015414798607a968ad04972.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/b9dfaa3e42c4430ca08d435780f70c57.png
[Spark_-----5.Spark task_Stage_ui_task_stage skipped_heayin123_-CSDN]: https://blog.csdn.net/u012684933/article/details/50378725
[4724dd85819641578b2b98e7fd17d4e5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/ac7772c977bc4b2fbd7c191d7c7f48e8.png
[fb59762062374a38984540c05613cd47.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/29bad6cd63cc466cb65bb5a3eddd31ee.png
[bd531600b5fc487abc7ef33bcc06539c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/4f1a926d8d5b436c8ef55a88ac630ee7.png
[6a1a00367c294ae6a7cfa2d50a65ec5f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/8d299e0deddf4ffe8c8258080225e4c0.png
[32d9a6e26a5040249576ce58db06a4ed.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/96a77ca725004ad18beb5bef7c4a2bfe.png
[6e7952d9409d4d65b8b404a6f2fe1a71.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/e1897974ae1e4e80927c9354df1678b2.png
[85c2bfaa9d514bbeb243c568a38c45be.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/fb0a8da18c3b4c76b7fadd1f238da8a4.png