缓存实现原理详解

缓存分为DiskStore（磁盘存储）和MemoryStore（内存存储）两种存储方式，下面我们通过Spark的源代码分别介绍它们如何实现Block（数据块）的存储和获取以及常用的缓存策略。

1.DiskStore如何存取Block（数据块）

（1）DiskStore（磁盘）可以配置多个folder（文件夹），Spark会在不同的folder下面创建Spark文件夹，文件夹的命名方式为（spark-local-yyyyMMddHHmmss-xxxx，xxxx是一个随机数），所有的block都会存储在Spark集群所创建的folder里面。DiskStore会在对象被创建时调用DiskBlockManager的createLocalDirs（conf）方法来创建文件夹。下面我们看一下具体的源码实现。

打开createLocalDirs（conf）方法，里面是创建文件夹的具体实现。

（2）在DiskStore里面，每一个block都被存储为一个file（文件），通过计算block id的hash值将block映射到文件中，block id与文件路径的映射关系如下所示（DiskBlockManager的getFile方法）：

从以上代码看出，传递每个block块的block id给getFile方法，然后调用Utils.nonNegative-Hash（filename）方法计算出hash值，将hash取模获得两个变量dirId和subDirId，在数组subDirs中找出相应的subDir，若没有则新建一个subDir，最后以subDir为路径、block id为文件名创建file handler（文件处理器）。

（3）DiskStore使用此file handler将block写入文件内，实现代码在DiskStore类的put-Bytes方法中，如下所示。

（4）获取block则非常简单，找到相应的文件并读取出来即可，实现代码在DiskStore类的getBytes方法中，如下所示。

在DiskStore中存取block首先是要将block id映射成相应的文件路径，接着存取文件就可以了。

2.MemoryStore如何存取Block（数据块）

（1）相对于DiskStore需要根据block Id的hash值来计算出文件路径并将block存放到对应的文件里面，MemoryStore管理block就显得非常简单：MemoryStore内部维护了一个hash map来管理所有的block，以block id为key将block存放到hash map中。block的内容被封装仅一个结构体MemoryEntry。

（2）在MemoryStore中存放block必须确保内存足够容纳下该block，JVM默认是60％可被内存缓存用来存储block，当存储的内容超过60％时，Spark会根据配置的缓存策略来决定是丢弃一些block还是将一些block存储到磁盘上。具体的实现代码在Spark源码的Memo-ryStore类的putArray方法中，如下所示。

（3）在MemoryStore类的tryToPut（）方法中，首先调用ensureFreeSpace（）方法确保空闲内存是否足以容纳block，若可以则将该block放入hash map中进行管理；若不足以容纳则通过调用dropFromMemory（）方法将block写入文件。

（4）从MemoryStore中取得block则非常简单，只需从hash map中取出block Id对应的value即可，具体实现代码在MemoryStore类的getValues（）方法中。

3.通过BlockManager来存取（Put or Get）Block（数据块）

上面介绍了DiskStore和MemoryStore对于block的存取操作，那么我们是要直接与它们交互存取数据吗，还是封装了更抽象的接口使我们无需关心底层？BlockManager为我们提供了doPut（）和get（）函数，用户可以使用这两个函数对block进行存取而无需关心底层实现。(www.xing528.com)

（1）对于doPut（）操作，主要分为以下3个步骤：为block创建BlockInfo结构体存储block相关信息，同时将其加锁使其不能被访问；根据block的storage level将block存储到内存或是磁盘上，同时解锁标识该block已经ready，可被访问；根据block的replication数决定是否将该block复制到远端。首先我们来看一下BlockManager的doPut（）函数的实现：

（2）接着我们来看一下BlockManager类的get（）方法的实现：

get（）方法首先会从本地（local）的BlockManager中查找block，如果找到则返回相应的block，若本地没有找到该block，则发起请求从其他的executor上的BlockManager中查找block。在通常情况下Spark任务的分配是根据block的分布决定的，任务往往会被分配到拥有block的结点上，因此getLocal（）就能找到所需的block；但是在资源有限的情况下，Spark会将任务调度到与block不同的结点上，这样就必须通过getRemote（）来获得block。

（3）我们继续跟踪这个get（）方法里对getLocal（）方法的调用，这个方法是从本地获取block，getLocal源代码实现如下：

在getLocal（）方法里会继续调用doGetLocal（）方法，在doGetLocal（）方法里首先会根据block id获得相应的BlockInfo并从中取出该block的storage level，根据storage level的不同，doGetLocal（）会从不同的缓存中取出数据。

（4）在BlockManager类的get（）方法里通过getRemote（）方法从远端获取block，ge-tRemote（）方法里面会调用doGetRemote（），在doGetRemote（）方法里首先取得该block的所有location信息，然后根据location向远端发送请求获取block，只要有一个远端返回block该函数就返回而不继续发送请求。接下来我们来看一下doGetRemote（）方法的具体实现：

至此我们简单介绍了如何通过BlockManager类中的get（）方法和put（）方法来轻易地存取block数据。

4.Partition和Block的关系

在Storage（存储管理）模块里面所有的操作都是和Block（数据块）相关的，但是在RDD里面所有的运算都是基于Partition（分区）的，那么Partition是如何与Block对应上的呢？在Spark中，Partition是一个逻辑上的概念，而Block是一个物理上的数据实体。一个RDD中的Partition对应于一个Storage模块中的Block。下面我们以RDD类的核心方法itera-tor（）方法为例分析一下Partition和Block的关系，以下是RDD类中iterator（）方法的实现：

在以上代码中，如果当前RDD的StorageLevel不是NONE的话，表示该RDD所包括的一系列Partition以Block的状态存储在BlockManager中，那么通过SparkEnv.get.cacheManager得到CacheManager对象，然后调用CacheManagerr中的getOrCompute（）方法计算RDD，在这个方法中Partition和Block发生了关系，Partition转化为Block，具体步骤如下：

（1）首先根据RDD id和Partition index构造出Block id（rdd_xx_xx），接着从BlockMan-ager中取出相应的Block。

（2）如果该Block存在，表示此RDD在之前已经被计算过和存储在BlockManager中，因此取出即可，无需再重新计算。

（3）如果该Block不存在则需要调用RDD的computeOrReadCheckpoint（）方法计算出新的Block，并将其存储到BlockManager中。

（4）需要注意的是Block的计算和存储是阻塞的，若另一线程也需要用到此Block则需等到该线程Block的装载结束。

这样RDD的transformation操作、action操作就和Block数据块建立了联系，虽然抽象上我们的操作是在Partition层面上进行的，但是Partition最终还是被映射成为Block，因此实际上我们的所有操作都是对Block的处理和存取。