码迷,mamicode.com
首页 > 其他好文 > 详细

Spark源码系列(五)RDD是如何被分布式缓存?

时间:2014-07-03 12:53:15      阅读:314      评论:0      收藏:0      [点我收藏+]

标签:des   style   blog   http   java   color   

这一章想讲一下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取数据。

bubuko.com,布布扣
  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 {
            loading.add(key)
          }
        }
        try {
          // 通过rdd自身的compute方法去计算得到结果,回去看看RDD那文章,自己看看源码就清楚了
          val computedValues = rdd.computeOrReadCheckpoint(split, context)

          // 如果是本地运行的,就没必要缓存了,直接返回即可
          if (context.runningLocally) {
            return computedValues
          }

          // 跟踪blocks的更新状态
          var updatedBlocks = Seq[(BlockId, BlockStatus)]()
          val returnValue: Iterator[T] = {
            if (storageLevel.useDisk && !storageLevel.useMemory) {
              /* 这是RDD采用DISK_ONLY的情况,直接扔给blockManager
               * 然后把结果直接返回,它不需要把结果一下子全部加载进内存
               * 这同样适用于MEMORY_ONLY_SER,但是我们需要在启用它之前确认blocks没被block store给丢弃 */
              updatedBlocks = blockManager.put(key, computedValues, storageLevel, tellMaster = true)
              blockManager.get(key) match {
                case Some(values) =>
                  values.asInstanceOf[Iterator[T]]
                case None =>
                  throw new Exception("Block manager failed to return persisted valued")
              }
            } else {
              // 先存到一个ArrayBuffer,然后一次返回,在blockManager里也存一份
              val elements = new ArrayBuffer[Any]
              elements ++= computedValues
              updatedBlocks = blockManager.put(key, elements, storageLevel, tellMaster = true)
              elements.iterator.asInstanceOf[Iterator[T]]
            }
          }

          // 更新task的监控参数
          val metrics = context.taskMetrics
          metrics.updatedBlocks = Some(updatedBlocks)

          new InterruptibleIterator(context, returnValue)

        } finally {
          // 改完了,释放锁
          loading.synchronized {
            loading.remove(key)
            loading.notifyAll()
          }
        }
    }
  }
View Code

1、如果blockManager当中有,直接从blockManager当中取。

2、如果blockManager没有,就先用RDD的compute函数得到出来一个Iterable接口。

3、如果StorageLevel是只保存在硬盘的话,就把值存在blockManager当中,然后从blockManager当中取,这样的好处是不会一次把数据全部加载进内存。

4、如果StorageLevel是需要使用内存的情况,就把结果添加到一个ArrayBuffer当中一次返回,另外在blockManager存上一份,下次直接从blockManager取。

对StorageLevel说明一下吧,贴一下它的源码。

bubuko.com,布布扣
class StorageLevel private(
    private var useDisk_ : Boolean,
    private var useMemory_ : Boolean,
    private var useOffHeap_ : Boolean,
    private var deserialized_ : Boolean,
    private var replication_ : Int = 1)

  val NONE = new StorageLevel(false, false, false, false)
  val DISK_ONLY = new StorageLevel(true, false, false, false)
  val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
  val MEMORY_ONLY = new StorageLevel(false, true, false, true)
  val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
  val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
  val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
  val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
  val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
  val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
  val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
  val OFF_HEAP = new StorageLevel(false, false, true, false)
View Code

大家注意看它那几个参数,useDisk_、useMemory_、useOffHeap_、deserialized_、replication_ 在具体的类型的时候是传的什么值。

下面我们的目标要放到blockManager。

BlockManager

BlockManager这个类比较大,我们从两方面开始看吧,putBytes和get方法。先从putBytes说起,之前说过Task运行结束之后,结果超过10M的话,会用BlockManager缓存起来。

env.blockManager.putBytes(blockId, serializedDirectResult, StorageLevel.MEMORY_AND_DISK_SER)

putBytes内部又掉了另外一个方法doPut,方法很大呀,先折叠起来。

bubuko.com,布布扣
  private def doPut(
      blockId: BlockId,
      data: Values,
      level: StorageLevel,
      tellMaster: Boolean = true): Seq[(BlockId, BlockStatus)] = {// Return value
    val updatedBlocks = new ArrayBuffer[(BlockId, BlockStatus)]

    // 记录它的StorageLevel,以便我们可以在它加载进内存之后,可以按需写入硬盘。
  // 此外,在我们把调用BlockInfo的markReay方法之前,都没法通过get方法获得该部分内容
    val putBlockInfo = {
      val tinfo = new BlockInfo(level, tellMaster)
      // 如果不存在,就添加到blockInfo里面
      val oldBlockOpt = blockInfo.putIfAbsent(blockId, tinfo)
      if (oldBlockOpt.isDefined) {
        // 如果已经存在了,就不需要重复添加了
        if (oldBlockOpt.get.waitForReady()) {return updatedBlocks
        }
        // 存在于blockInfo当中->但是上一次保存失败了,拿出旧的信息,再试一遍
        oldBlockOpt.get
      } else {
        tinfo
      }
    }

    val startTimeMs = System.currentTimeMillis
    // 当我们需要存储数据,并且要复制数据到别的机器,我们需要访问它的值,但是因为我们的put操作会读取整个iterator,
    // 这就不会有任何的值留下。在我们保存序列化的数据的场景,我们可以记住这些bytes,但在其他场景,比如反序列化存储的
    // 时候,我们就必须依赖返回一个Iterator
    var valuesAfterPut: Iterator[Any] = null
    // Ditto for the bytes after the put
    var bytesAfterPut: ByteBuffer = null
    // Size of the block in bytes
    var size = 0L

    // 在保存数据之前,我们要实例化,在数据已经序列化并且准备好发送的情况下,这个过程是很快的
    val replicationFuture = if (data.isInstanceOf[ByteBufferValues] && level.replication > 1) {
      // duplicate并不是复制这些数据,只是做了一个包装
      val bufferView = data.asInstanceOf[ByteBufferValues].buffer.duplicate()
      Future {
        // 把block复制到别的机器上去
        replicate(blockId, bufferView, level)
      }
    } else {
      null
    }

    putBlockInfo.synchronized {

      var marked = false
      try {
        if (level.useMemory) {
          // 首先是保存到内存里面,尽管它也使用硬盘,等内存不够的时候,才会写入硬盘
          // 下面分了三种情况,但是Task的结果是ByteBufferValues这种情况,具体看putBytes方法
          val res = data match {
            case IteratorValues(iterator) =>
              memoryStore.putValues(blockId, iterator, level, true)
            case ArrayBufferValues(array) =>
              memoryStore.putValues(blockId, array, level, true)
            case ByteBufferValues(bytes) =>
              bytes.rewind()
              memoryStore.putBytes(blockId, bytes, level)
          }
          size = res.size
          // 这里写得那么恶心,是跟data的类型有关系的,data: Either[Iterator[_], ByteBuffer],Left是Iterator,Right是ByteBuffer
          res.data match {
            case Right(newBytes) => bytesAfterPut = newBytes
            case Left(newIterator) => valuesAfterPut = newIterator
          }
          // 把被置换到硬盘的blocks记录到updatedBlocks上
          res.droppedBlocks.foreach { block => updatedBlocks += block }
        } else if (level.useOffHeap) {
          // 保存到Tachyon上.
          val res = data match {
            case IteratorValues(iterator) =>
              tachyonStore.putValues(blockId, iterator, level, false)
            case ArrayBufferValues(array) =>
              tachyonStore.putValues(blockId, array, level, false)
            case ByteBufferValues(bytes) =>
              bytes.rewind()
              tachyonStore.putBytes(blockId, bytes, level)
          }
          size = res.size
          res.data match {
            case Right(newBytes) => bytesAfterPut = newBytes
            case _ =>
          }
        } else {
          // 直接保存到硬盘,不要复制到其它节点的就别返回数据了.
          val askForBytes = level.replication > 1
          val res = data match {
            case IteratorValues(iterator) =>
              diskStore.putValues(blockId, iterator, level, askForBytes)
            case ArrayBufferValues(array) =>
              diskStore.putValues(blockId, array, level, askForBytes)
            case ByteBufferValues(bytes) =>
              bytes.rewind()
              diskStore.putBytes(blockId, bytes, level)
          }
          size = res.size
          res.data match {
            case Right(newBytes) => bytesAfterPut = newBytes
            case _ =>
          }
        }
     // 通过blockId获得当前的block状态
        val putBlockStatus = getCurrentBlockStatus(blockId, putBlockInfo)
        if (putBlockStatus.storageLevel != StorageLevel.NONE) {
          // 成功了,把该block标记为ready,通知BlockManagerMaster
          marked = true
          putBlockInfo.markReady(size)
          if (tellMaster) {
            reportBlockStatus(blockId, putBlockInfo, putBlockStatus)
          }
          updatedBlocks += ((blockId, putBlockStatus))
        }
      } finally {
        // 如果没有标记成功,就把该block信息清除
       if (!marked) {
          blockInfo.remove(blockId)
          putBlockInfo.markFailure()
        }
      }
    }

    // 把数据发送到别的节点做备份
    if (level.replication > 1) {
      data match {
        case ByteBufferValues(bytes) => Await.ready(replicationFuture, Duration.Inf)
        case _ => {
          val remoteStartTime = System.currentTimeMillis
          // 把Iterator里面的数据序列化之后,发送到别的节点
          if (bytesAfterPut == null) {
            if (valuesAfterPut == null) {
              throw new SparkException("Underlying put returned neither an Iterator nor bytes! This shouldn‘t happen.")
            }
            bytesAfterPut = dataSerialize(blockId, valuesAfterPut)
          }
          replicate(blockId, bytesAfterPut, level)
        }
      }
    }
    // 销毁bytesAfterPut
    BlockManager.dispose(bytesAfterPut)
    updatedBlocks
  }
View Code

从上面的的来看:

1、存储的时候按照不同的存储级别分了3种情况来处理:存在内存当中(包括MEMORY字样的),存在tachyon上(OFF_HEAP),只存在硬盘上(DISK_ONLY)。

2、存储完成之后会根据存储级别决定是否发送到别的节点,在名字上最后带2字的都是这种,2表示一个block会在两个节点上保存。

3、存储完毕之后,会向BlockManagerMaster汇报block的情况。

4、这里面的序列化其实是先压缩后序列化,默认使用的是LZF压缩,可以通过spark.io.compression.codec设定为snappy或者lzo,序列化方式通过spark.serializer设置,默认是JavaSerializer。

bubuko.com,布布扣

接下来我们再看get的情况。

    val local = getLocal(blockId)
    if (local.isDefined) return local
    val remote = getRemote(blockId)
    if (remote.isDefined) return remote
    None

先从本地取,本地没有再去别的节点取,都没有,返回None。从本地取就不说了,怎么进怎么出。讲一下怎么从别的节点去,它们是一个什么样子的关系?

我们先看getRemote方法

bubuko.com,布布扣
  private def doGetRemote(blockId: BlockId, asValues: Boolean): Option[Any] = {
    val locations = Random.shuffle(master.getLocations(blockId))
    for (loc <- locations) {
      val data = BlockManagerWorker.syncGetBlock(GetBlock(blockId), ConnectionManagerId(loc.host, loc.port))
      if (data != null) {
        if (asValues) {
          return Some(dataDeserialize(blockId, data))
        } else {
          return Some(data)
        }
      }
    }
    None
  }
View Code

这个方法包括两个步骤:

1、用blockId通过master的getLocations方法找到它的位置。

2、通过BlockManagerWorker.syncGetBlock到指定的节点获取数据。

ok,下面就重点讲BlockManager和BlockManagerMaster之间的关系,以及BlockManager之间是如何相互传输数据。

BlockManager与BlockManagerMaster的关系

BlockManager我们使用的时候是从SparkEnv.get获得的,我们观察了一下SparkEnv,发现它包含了我们运行时候常用的那些东东。那它创建是怎么创建的呢,我们找到SparkEnv里面的create方法,右键FindUsages,就会找到两个地方调用了,一个是SparkContext,另一个是Executor。在SparkEnv的create方法里面会实例化一个BlockManager和BlockManagerMaster。这里我们需要注意看BlockManagerMaster的实例化方法,里面调用了registerOrLookup方法。

    def registerOrLookup(name: String, newActor: => Actor): ActorRef = {
      if (isDriver) {
        actorSystem.actorOf(Props(newActor), name = name)
      } else {
        val driverHost: String = conf.get("spark.driver.host", "localhost")
        val driverPort: Int = conf.getInt("spark.driver.port", 7077)
        Utils.checkHost(driverHost, "Expected hostname")
        val url = s"akka.tcp://spark@$driverHost:$driverPort/user/$name"
        val timeout = AkkaUtils.lookupTimeout(conf)
        Await.result(actorSystem.actorSelection(url).resolveOne(timeout), timeout)
      }
    }

所以从这里可以看出来,除了Driver之后的actor都是,都是持有的Driver的引用ActorRef。梳理一下,我们可以得出以下结论:

1、SparkContext持有一个BlockManager和BlockManagerMaster。

2、每一个Executor都持有一个BlockManager和BlockManagerMaster。

3、Executor和SparkContext的BlockManagerMaster通过BlockManagerMasterActor来通信。

接下来,我们看看BlockManagerMasterActor里的三组映射关系。

  // 1、BlockManagerId和BlockManagerInfo的映射关系
  private val blockManagerInfo = new mutable.HashMap[BlockManagerId, BlockManagerInfo]
  // 2、Executor ID 和 Block manager ID的映射关系
  private val blockManagerIdByExecutor = new mutable.HashMap[String, BlockManagerId]
  // 3、BlockId和保存它的BlockManagerId的映射关系
  private val blockLocations = new JHashMap[BlockId, mutable.HashSet[BlockManagerId]]

看到这三组关系,前面的getLocations方法不用看它的实现,我们都应该知道是怎么找了。

BlockManager相互传输数据

BlockManager之间发送数据和接受数据是通过BlockManagerWorker的syncPutBlock和syncGetBlock方法来实现。看BlockManagerWorker的注释,说是BlockManager的网络接口,采用的是事件驱动模型。

再仔细看这两个方法,它传输的数据包装成BlockMessage之后,通过ConnectionManager的sendMessageReliablySync方法来传输。

接下来的故事就是nio之间的发送和接收了,就简单说几点吧:

1、ConnectionManager内部实例化一个selectorThread线程来接收小心,具体请看run方法。

2、Connection发送数据的时候,是一次把消息队列的message全部发送,不是一个一个message发送,具体看SendConnection的write方法,与之对应的接收看ReceivingConnection的read方法。

3、read完了之后,调用回调函数ConnectionManager的receiveMessage方法,它又调用了handleMessage方法,handleMessage又调用了BlockManagerWorker的onBlockMessageReceive方法。剧情有点儿狗血,我是快晕了。

  def processBlockMessage(blockMessage: BlockMessage): Option[BlockMessage] = {
    blockMessage.getType match {
      case BlockMessage.TYPE_PUT_BLOCK => {
        val pB = PutBlock(blockMessage.getId, blockMessage.getData, blockMessage.getLevel)
        putBlock(pB.id, pB.data, pB.level)
        None
      }
      case BlockMessage.TYPE_GET_BLOCK => {
        val gB = new GetBlock(blockMessage.getId)
        val buffer = getBlock(gB.id)
        Some(BlockMessage.fromGotBlock(GotBlock(gB.id, buffer)))
      }
      case _ => None
    }
  }

根据BlockMessage的类型进行处理,put类型就保存数据,get类型就从本地把block读出来返回给它。

bubuko.com,布布扣

相关参数

//BlockManager的最大内存
spark.storage.memoryFraction 默认值0.6
//文件保存的位置
spark.local.dir 默认是系统变量java.io.tmpdir的值
//tachyon保存的地址
spark.tachyonStore.url 默认值tachyon://localhost:19998
//默认不启用netty来传输shuffle的数据
spark.shuffle.use.netty 默认值是false
spark.shuffle.sender.port 默认值是0
// Max megabytes of data to keep in flight per reducer (to avoid over-allocating memory for receiving shuffle outputs)
spark.reducer.maxMbInFlight 默认值是48*1024*1024

 

岑玉海

转载请注明出处,谢谢!

 

Spark源码系列(五)RDD是如何被分布式缓存?,布布扣,bubuko.com

Spark源码系列(五)RDD是如何被分布式缓存?

标签:des   style   blog   http   java   color   

原文地址:http://www.cnblogs.com/cenyuhai/p/3808774.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!