Spark源码系列（五）分布式缓存源码系列（五）分布式缓存
这一章想讲一下Spark的缓存是如何实现的。这个persist方法是在RDD里面的，所以我们直接打开RDD这个类。
def persist(newLevel: StorageLevel): this.type = {
// StorageLevel不能随意更改
if (storageLevel != StorageLevel.NONE && newLevel != storageLevel) {
throw new UnsupportedOperationException("Cannot change storage level of an RDD after it was already assigned a level")
}
sc.persistRDD(this)
// Register the RDD with the ContextCleaner for automatic GC-based cleanup
// 注册清理方法
sc.cleaner.foreach(_.registerRDDForCleanup(this))
storageLevel = newLevel
this
}
它调用SparkContext去缓存这个RDD，追杀下去。
private[spark] def persistRDD(rdd: RDD[_]) {
persistentRdds(rdd.id) = rdd
}
它居然是用一个HashMap来存的，具体看这个map的类型是TimeStampedWeakValueHashMap[Int, RDD[_]]类型。把存进去
的值都隐式转换成WeakReference，然后加到一个内部的一个ConcurrentHashMap里面。这里貌似也没干啥，这是有个鸟蛋
用。。大神莫喷，知道干啥用的人希望告诉我一下。
CacheManager
现在并没有保存，等到真正运行Task运行的时候才会去缓存起来。入口在Task的runTask方法里面，具体的我们可以看
ResultTask，它调用了RDD的iterator方法。
final def iterator(split: Partition, context: TaskContext): Iterator[T] = {
if (storageLevel != StorageLevel.NONE) {
SparkEnv.get.cacheManager.getOrCompute(this, split, context, storageLevel)
} else {
computeOrReadCheckpoint(split, context)
}
}
一旦设置了StorageLevel，就要从SparkEnv的cacheManager取数据。
def getOrCompute[T](rdd: RDD[T], split: Partition, context: TaskContext, storageLevel: StorageLevel): Iterator[T] = {
val key = RDDBlockId(rdd.id, split.index)
blockManager.get(key) match {
case Some(values) =>
// 已经有了，直接返回就可以了
new InterruptibleIterator(context, values.asInstanceOf[Iterator[T]])
case None =>
// loading包含这个key表示已经有人在加载了，等到loading被释放了，就可以去blockManager里
面取到了
loading.synchronized {
if (loading.contains(key)) {
while (loading.contains(key)) {
try {
loading.wait()
} catch {
case e: Exception =>
logWarning(s"Got an exception while waiting for another thread to load $key", e)
}
}
// 别人成功拿到了，我们直接取结果就是了，如果别人取失败了，我们再来取一次
blockManager.get(key) match {
case Some(values) =>
return new InterruptibleIterator(context, values.asInstanceOf[Iterator[T]])
case None =>
loading.add(key)
}
} else {
剩余6页未读，继续阅读
资源评论
