spark存储管理之磁盘存储--DiskStore

diskstore

接着上一篇，本篇，我们分析一下实现磁盘存储的功能类diskstore，这个类相对简单。在正式展开之前，我觉得有必要大概分析一下blockmanager的背景，或者说它的运行环境，运行的作用范围。blockmanager这个类其实在运行时的每个节点都会有一个实例（包括driver和executor进程），因为不论是driver端进行广播变量的创建，还是executor端shuffle过程中写shuffle块，或者是任务运行时结果太大需要通过blockmanager传输，或者是rdd的缓存，其实在每个运行节点上都会通过blockmanager来管理程序内部对于本地的内存和磁盘的读写，所以综上，我想表达的核心意思就是每个进程（driver和executor）都有一blockmanager实例，而这些blockmanager实例是通过blockmanagerid类来进行唯一区分的，blockmanagerid实际上是对进程物理位置的封装。

diskstore.put

首先我们来看一个最常用的写入方法

def put(blockid: blockid)(writefunc: writablebytechannel => unit): unit = {
    // 通过diskblockmanager对象检查这个blockid对应的文件名的文件是否存在
    if (contains(blockid)) {
      throw new illegalstateexception(s"block $blockid is already present in the disk store")
    }
    logdebug(s"attempting to put block $blockid")
    val starttime = system.currenttimemillis
    // 通过diskblockmanager获取一个文件用于写入数据
    val file = diskmanager.getfile(blockid)
    // 用countingwritablechannel包装一下，以便于记录写入的字节数
    val out = new countingwritablechannel(openforwrite(file))
    var threwexception: boolean = true
    try {
      writefunc(out)
      // 关键步骤，记录到内部的map结构中
      blocksizes.put(blockid, out.getcount)
      threwexception = false
    } finally {
      try {
        out.close()
      } catch {
        case ioe: ioexception =>
          if (!threwexception) {
            threwexception = true
            throw ioe
          }
      } finally {
         if (threwexception) {
          remove(blockid)
        }
      }
    }
    val finishtime = system.currenttimemillis
    logdebug("block %s stored as %s file on disk in %d ms".format(
      file.getname,
      utils.bytestostring(file.length()),
      finishtime - starttime))
  }

这个方法很简单，没什么好说的，但是调用了一个比较重要的类diskblockmanager，这个类的功能就是对磁盘上的目录和文件进行管理，会在磁盘上按照一定规则创建一些目录和子目录，在分配文件名时也会尽量均匀第分配在这些目录和子目录下。

diskstore.putbytes

这个方法就不说了，简单处理一下直接调用put方法。

  def putbytes(blockid: blockid, bytes: chunkedbytebuffer): unit = {
    put(blockid) { channel =>
      bytes.writefully(channel)
    }
  }

diskstore.getbytes

我们来看一下这个方法，首先通过diskblockmanager获取对应的文件名，然后将其包装成一个blockdata对象，分为加密和不加密两种。

  def getbytes(blockid: blockid): blockdata = {
    val file = diskmanager.getfile(blockid.name)
    val blocksize = getsize(blockid)
  
    securitymanager.getioencryptionkey() match {
      case some(key) =>
        // encrypted blocks cannot be memory mapped; return a special object that does decryption
        // and provides inputstream / fileregion implementations for reading the data.
        new encryptedblockdata(file, blocksize, conf, key)
  
      case _ =>
        // 看一下diskblockdata
        new diskblockdata(minmemorymapbytes, maxmemorymapbytes, file, blocksize)
    }
  }

diskblockdata

这个类作为磁盘文件的包装类，主要功能是提供了几个方便的接口，将磁盘文件中的数据读取出来并生成缓冲对象。
这个类中有两个重要的方法tochunkedbytebuffer和tobytebuffer，tobytebuffer就不说了，调用readablebytechannel.read(bytebuffer dst)方法读取文件数据，我们看一下tochunkedbytebuffer

diskblockdata.tochunkedbytebuffer

这个方法也很简单，在数据量比较大的时候，由于每次申请的内存块大小有限制maxmemorymapbytes，所以需要切分成多个块

  override def tochunkedbytebuffer(allocator: (int) => bytebuffer): chunkedbytebuffer = {
    // utils.trywithresource调用保证在使用完资源后关闭资源
    // 基本等同于java中的try{}finally{}
    utils.trywithresource(open()) { channel =>
      var remaining = blocksize
      val chunks = new listbuffer[bytebuffer]()
      while (remaining > 0) {
        // 这里取剩余大小和maxmemorymapbytes的较小值，
        // 也就是说每次申请的内存块大小不超过maxmemorymapbytes
        val chunksize = math.min(remaining, maxmemorymapbytes)
        val chunk = allocator(chunksize.toint)
        remaining -= chunksize
        javautils.readfully(channel, chunk)
        chunk.flip()
        chunks += chunk
      }
      new chunkedbytebuffer(chunks.toarray)
    }
  }

diskblockmanager

这个类之前也分析过，主要是用来管理spark运行过程中写入的一些临时文件，以及目录的管理。

• 首先会根据参数配置创建本地目录（可以是逗号分隔的多个目录），参数的优先顺序是：如果是运行在yarn上，则会使用yarn参数local_dirs配置的本地目录；否则获取环境变量spark_local_dirs的值；否则获取spark.local.dir参数的值；最后如果都没有配置，那么就用java系统参数java.io.tmpdir的值作为临时目录。

• 其次，关于文件在目录之间分配的问题，使用文件名的hash值对目录数量取余的方法来尽量将文件均匀地分配到不同的目录下。

• 另外一点要说的是文件名的命名规则，是根据不同作用的block来区别命名的，例如rdd缓存写入的block的id就是rddblockid，它的文件名拼接规则是"rdd_" + rddid + "_" + splitindex