ExternalBlockStore

Posted 2022-11-21 大冰的小屋

ExternalBlockStore作为存储Block文件的外部管理器，其实际是封装了TachyonBlockManager。这样做可以更好的解耦合，方便以后加入新的第三方的管理器。ExternalBlockStore创建TachyonBlockManager代码如下：

/**
 * Stores BlockManager blocks on ExternalBlockStore.
 * We capture any potential exception from underlying implementation
 * and return with the expected failure value
 */
private[spark] class ExternalBlockStore(blockManager: BlockManager, executorId: String)
  extends BlockStore(blockManager: BlockManager) with Logging 

  lazy val externalBlockManager: Option[ExternalBlockManager] = createBlkManager()

  ...

  // Create concrete block manager and fall back to Tachyon by default for backward compatibility.
  private def createBlkManager(): Option[ExternalBlockManager] = 
    val clsName = blockManager.conf.getOption(ExternalBlockStore.BLOCK_MANAGER_NAME)
      .getOrElse(ExternalBlockStore.DEFAULT_BLOCK_MANAGER_NAME)

    try 
      val instance = Utils.classForName(clsName)
        .newInstance()
        .asInstanceOf[ExternalBlockManager]
      instance.init(blockManager, executorId)
      ShutdownHookManager.addShutdownHook  () =>
        logDebug("Shutdown hook called")
        externalBlockManager.map(_.shutdown())
      
      Some(instance)
     catch 
      case NonFatal(t) =>
        logError("Cannot initialize external block store", t)
        None
    
  

...

private[spark] object ExternalBlockStore extends Logging 
  val MAX_DIR_CREATION_ATTEMPTS = 10
  val SUB_DIRS_PER_DIR = "64"
  val BASE_DIR = "spark.externalBlockStore.baseDir"
  val FOLD_NAME = "spark.externalBlockStore.folderName"
  val MASTER_URL = "spark.externalBlockStore.url"
  val BLOCK_MANAGER_NAME = "spark.externalBlockStore.blockManager"
  val DEFAULT_BLOCK_MANAGER_NAME = "org.apache.spark.storage.TachyonBlockManager"

Tachyon是什么

Tachyon是AMPLab开发的一款内存分布式文件系统。它介于计算层和存储层之间，可以简单的理解为存储层在内存内的一个Cache系统。作为以内存为中心的高容错的分布式文件系统，能够为集群（例如Spark、Map-Reduce等）提供可靠的内存级的文件共享服务。 同Spark和Hadoop一样，Tachyon是完全开源的，并且也是一个以JVM为base的系统。Tachyon和Spark都出自AMPlab，Tachyon作为了Spark默认的off-heap内存存储框架。

为什么要选择使用Tachyon

原因如下：
1. Spark的ShuffleMapTask和ResultTask被划分到不同的Stage，ShuffleMapTask执行完毕将中间结果输出到本地磁盘文件系统（如HDFS），然后下一Stage中的ResultTask通过shuffleClient下载ShuffleMapTask的输出到本地磁盘文件系统，这种基于磁盘的读写效率较低；
2. Spark的计算引擎与存储体系都位于Executor的同一个进程中，当计算执行崩溃出错后，存储体系缓存的数据也会全部丢失，这时Spark就不得不根据数据的Lineage，重新就算丢失的数据；
3. 不同的Spark任务可能会访问同样的数据，例如两个任务都要访问HDFS中的某些Block，每个任务都要自己去磁盘加载数据到内存中。这导致了数据被重复加载到内存，数据对象太多导致Java GC时间过长等问题。