前言

HDFS内存存储原理

• 第一种,数据就是临时维持在内存中的,服务一停止,数据全没.
• 第二种,数据还是存在与内存种,但是服务停止的时候,做持久化处理,最终全部写入到磁盘.

仔细来看以上这2种观点,其实都有不小的瑕疵.

首先第一个观点,服务一旦停止,内存数据全丢,这个是无法接受的,我们可以忍受内存中少量的数据丢失,但是全丢就不是特别好的处理方式了.而且这个也有点不合理,内存的存储空间是有限的,如果不及时存储一部分数据,内存空间迟早会耗尽.

然后是第二个观点,第二个方案种是在服务停止退出的时候做持久化操作,但是他同样会面临上面提到的内存空间的限制问题.而且假设机器的内存是足够大的,那么最后写入磁盘的那个阶段想必也不会那么快,因为数据可能会很多.

所以一般的通用的比较好的做法是异步的做持久化,什么意思呢

内存存储新数据的同时,持久化距离当前时刻最远(存储时间最早)的数据

换一个通俗的解释,好比我有个内存数据块队列,在队列头部不断有新增的数据块插入,就是待存储的块,因为资源有限 ,我要把队列尾部的块,也就是早些时间点的块持久化到磁盘中,然后才有空间腾出来存新的块.然后形成这样的一个循环,新的块加入,老的块移除,保证了整体数据的更新.

HDFS的LAZY_PERSIST内存存储策略用的就是这套方法.下面是一张原理图:

上文描述的原理在图中的表示其实是4,6,的步骤.写数据的RAM,然后异步的写到Disk.前面几个步骤是如何设置StorageType的操作,这个在下文种会具体提到.所以上图所示的大体步骤可以归纳为如下:

• 第一步,对目标文件目录设置StoragePolicy为LAZY_PERSIST的内存存储策略.
• 第二步,客户端进程向NameNode发起创建/写文件的请求.
• 第三步,然后请求到具体的DataNode,DataNode会把这些数据块写入RAM内存中,同时启动异步线程服务将内存数据持久化到磁盘上.

内存的异步持久化存储,就是明显不同于其他介质存储数据的地方.这应该也是LAZY_PERSIST的名称的源由吧,数据不是马上落盘,而是”lazy persisit”懒惰的方式,延时的处理.

Linux 虚拟内存盘

HDFS的内存存储流程分析

HDFS文件内存存储设置

• 第一种,通过命令行的方式,调用如下命令

hdfs storagepolicies -setStoragePolicy -path <path> -policy LAZY_PERSIST

方便,快速.

• 第二种,通过调用对应的程序方法,比如调用暴露到外部的create文件方法,但是得带上参数CreateFlag.LAZY_PERSIST.例子如下:

FSDataOutputStream fos =
        fs.create(
            path,
            FsPermission.getFileDefault(),
            EnumSet.of(CreateFlag.CREATE, CreateFlag.LAZY_PERSIST),
            bufferLength,
            replicationFactor,
            blockSize,
            null);

上述方式最终调用的是DFSClient的create同名方法,如下:

  /**
   * Call {@link #create(String, FsPermission, EnumSet, boolean, short,
   * long, Progressable, int, ChecksumOpt)} with <code>createParent</code>
   *  set to true.
   */
  public DFSOutputStream create(String src, FsPermission permission,
      EnumSet<CreateFlag> flag, short replication, long blockSize,
      Progressable progress, int buffersize, ChecksumOpt checksumOpt)
      throws IOException {
    return create(src, permission, flag, true,
        replication, blockSize, progress, buffersize, checksumOpt, null);
  }

方法经过RPC层层调用,经过FSNamesystem,最终会到FSDirWriteFileOp的startFile方法,在此方法内部,会有设置的动作

  static HdfsFileStatus startFile(
      FSNamesystem fsn, FSPermissionChecker pc, String src,
      PermissionStatus permissions, String holder, String clientMachine,
      EnumSet<CreateFlag> flag, boolean createParent,
      short replication, long blockSize,
      EncryptionKeyInfo ezInfo, INode.BlocksMapUpdateInfo toRemoveBlocks,
      boolean logRetryEntry)
      throws IOException {
    assert fsn.hasWriteLock();

    boolean create = flag.contains(CreateFlag.CREATE);
    boolean overwrite = flag.contains(CreateFlag.OVERWRITE);
    // 判断CreateFlag是否带有LAZY_PERSIST标识,来判断是否是内存存储策略的
    boolean isLazyPersist = flag.contains(CreateFlag.LAZY_PERSIST);

    ...
    // 然后在此设置策略
    setNewINodeStoragePolicy(fsd.getBlockManager(), newNode, iip,
                             isLazyPersist);
    fsd.getEditLog().logOpenFile(src, newNode, overwrite, logRetryEntry);
    if (NameNode.stateChangeLog.isDebugEnabled()) {
      NameNode.stateChangeLog.debug("DIR* NameSystem.startFile: added " +
          src + " inode " + newNode.getId() + " " + holder);
    }
    return FSDirStatAndListingOp.getFileInfo(fsd, src, false, isRawPath);
  }

所以这部分的过程调用图如下:

OK,以上就是前期存储策略的设置过程了,这一部分还是非常的直接明了的.

LAZY_PERSIST内存存储

LAZY_PERSIST相关结构

在FsDatasetImpl中,与内存存储相关的服务对象有如下的3个.

下面来一个个介绍:

LazyWriter:lazyWriter是一个线程服务,此线程会不断的循环着从数据块列表中取出数据块,加入到异步持久化线程池RamDiskAsyncLazyPersistService中去执行.

RamDiskAsyncLazyPersistService:此对象就是异步持久化线程服务,里面针对每一个磁盘块设置一个对应的线程池,然后需要持久化到给定的磁盘块的数据块会被提交到对应的线程池中去.每个线程池的最大线程数为1.

RamDiskReplicaLruTracker:副本块跟踪类,此类种维护了所有已持久化,未持久化的副本以及总副本数据信息.所以当一个副本被最终存储到内存种后,相应的会有副本所属队列信息的变更.其次当节点内存不足的时候,部分距离最近最久没有被访问的副本块会在此类中被移除.

综合了以上3者的紧密合作,最终实现了HDFS的内存存储.下面是具体的角色介绍.

RamDiskReplicaLruTracker

public class RamDiskReplicaLruTracker extends RamDiskReplicaTracker {
  ...
  /**
   * Map of blockpool ID to <map of blockID to ReplicaInfo>.
   */
  Map<String, Map<Long, RamDiskReplicaLru>> replicaMaps;

  /**
   * Queue of replicas that need to be written to disk.
   * Stale entries are GC‘d by dequeueNextReplicaToPersist.
   */
  Queue<RamDiskReplicaLru> replicasNotPersisted;

  /**
   * Map of persisted replicas ordered by their last use times.
   */
  TreeMultimap<Long, RamDiskReplicaLru> replicasPersisted;
  ...

这里的Queue就是待内存存储队列.以上3个变量之间的关系图如下

RamDiskReplicaLruTracker中的方法操作绝大多数与这3个变量的增删改动相关,所以逻辑并不复杂,我们只需要了解这些方法有什么作用即可.我对此分成了2类:

第一类,异步持久化操作相关方法.如图:

当节点重启或者有新的文件被设置了LAZY_PERSIST策略后,就会有新的副本块被存储到内存中,同时会加入到replicaNotPersisted队列中.然后经过中间的dequeueNextReplicaToPersist取出下一个将被持久化的副本块,进行写磁盘的操作.recordStartLazyPersist,recordEndLazyPersist这2个方法会在持久化的过程中被调用,标志着持久化状态的变更.

另一类,异步持久化操作无直接关联方法.如图:

有下面3个方法:

• discardReplica:当此副本已经被检测出不需要的时候,包括已被删除,或已损坏的情况,可以从内存中移除,撤销.
• touch:恰好与Linux种的touch同名,此方法意味访问了一次某特定的副本块,并会更新此副本块的lastUesdTime. lastUesdTime会在后面提到的LRU算法中起到关键的作用.
• getNextCandidateForEviction:此方法在DataNode内存空间不足,需要内存额外预留出空间给新的副本块存放时被调用.此方法会根据所设置的eviction scheme模式,选择需要被移除的块,默认的是LRU策略的.

这里反复提到一个名词,LRU,他的全称是Least Recently Used,意为最近最少使用算法,相关链接点此,getNextCandidateForEviction采用此算法的好处是保证了现有副本块的一个活跃度,把最近很久没有访问过的给移除掉.对于这个操作,我们有必要了解其中的细节.

先是touch会更新最近访问的时间

  synchronized void touch(final String bpid,
                          final long blockId) {
    Map<Long, RamDiskReplicaLru> map = replicaMaps.get(bpid);
    RamDiskReplicaLru ramDiskReplicaLru = map.get(blockId);

    ...

    // Reinsert the replica with its new timestamp.
    // 更新最近访问时间戳,并重新插入数据
    if (replicasPersisted.remove(ramDiskReplicaLru.lastUsedTime, ramDiskReplicaLru)) {
      ramDiskReplicaLru.lastUsedTime = Time.monotonicNow();
      replicasPersisted.put(ramDiskReplicaLru.lastUsedTime, ramDiskReplicaLru);
    }
  }

然后是第二步获取候选移除块

  synchronized RamDiskReplicaLru getNextCandidateForEviction() {
    // 获取replicasPersisted迭代器进行遍历
    final Iterator<RamDiskReplicaLru> it = replicasPersisted.values().iterator();
    while (it.hasNext()) {
      // 因为replicasPersisted已经根据时间排好序了,所以取出当前的块进行移除即可
      final RamDiskReplicaLru ramDiskReplicaLru = it.next();
      it.remove();

      Map<Long, RamDiskReplicaLru> replicaMap =
          replicaMaps.get(ramDiskReplicaLru.getBlockPoolId());

      if (replicaMap != null && replicaMap.get(ramDiskReplicaLru.getBlockId()) != null) {
        return ramDiskReplicaLru;
      }

      // The replica no longer exists, look for the next one.
    }
    return null;
  }

这里比较有意思的是,根据已持久化的块的访问时间来进行筛选移除,而不是直接是内存中的块.最后是在内存中移除与候选块属于同一副本信息的块并释放内存空间.

    /**
     * Attempt to evict one or more transient block replicas until we
     * have at least bytesNeeded bytes free.
     */
    public void evictBlocks(long bytesNeeded) throws IOException {
      int iterations = 0;

      final long cacheCapacity = cacheManager.getCacheCapacity();
      // 当检测到内存空间不满足外界需要的大小时
      while (iterations++ < MAX_BLOCK_EVICTIONS_PER_ITERATION &&
             (cacheCapacity - cacheManager.getCacheUsed()) < bytesNeeded) {
        // 获取待移除副本信息
        RamDiskReplica replicaState = ramDiskReplicaTracker.getNextCandidateForEviction();

        if (replicaState == null) {
          break;
        }

        if (LOG.isDebugEnabled()) {
          LOG.debug("Evicting block " + replicaState);
        }

        ...
        // 移除内存中的相关块并释放空间
          // Delete the block+meta files from RAM disk and release locked
          // memory.
          removeOldReplica(replicaInfo, newReplicaInfo, blockFile, metaFile,
              blockFileUsed, metaFileUsed, bpid);
        }
      }
    }

LazyWriter

    public void run() {
      int numSuccessiveFailures = 0;

      while (fsRunning && shouldRun) {
        try {
          // 取出新的副本块并提交到异步服务中,返回是否提交成功布尔值
          numSuccessiveFailures = saveNextReplica() ? 0 : (numSuccessiveFailures + 1);

          // Sleep if we have no more work to do or if it looks like we are not
          // making any forward progress. This is to ensure that if all persist
          // operations are failing we don‘t keep retrying them in a tight loop.
          if (numSuccessiveFailures >= ramDiskReplicaTracker.numReplicasNotPersisted()) {
            Thread.sleep(checkpointerInterval * 1000);
            numSuccessiveFailures = 0;
          }
        } catch (InterruptedException e) {
          LOG.info("LazyWriter was interrupted, exiting");
          break;
        } catch (Exception e) {
          LOG.warn("Ignoring exception in LazyWriter:", e);
        }
      }
    }

进入saveNextReplica方法的处理

    private boolean saveNextReplica() {
      RamDiskReplica block = null;
      FsVolumeReference targetReference;
      FsVolumeImpl targetVolume;
      ReplicaInfo replicaInfo;
      boolean succeeded = false;

      try {
        // 从队列种取出新的待持久化的块
        block = ramDiskReplicaTracker.dequeueNextReplicaToPersist();
        if (block != null) {
          synchronized (FsDatasetImpl.this) {
            ...
            // 提交到异步服务中去
              asyncLazyPersistService.submitLazyPersistTask(
                  block.getBlockPoolId(), block.getBlockId(),
                  replicaInfo.getGenerationStamp(), block.getCreationTime(),
                  replicaInfo.getMetaFile(), replicaInfo.getBlockFile(),
                  targetReference);
            }
          }
        }
        succeeded = true;
      } catch(IOException ioe) {
        LOG.warn("Exception saving replica " + block, ioe);
      } finally {
        if (!succeeded && block != null) {
          LOG.warn("Failed to save replica " + block + ". re-enqueueing it.");
          onFailLazyPersist(block.getBlockPoolId(), block.getBlockId());
        }
      }
      return succeeded;
    }

所以LazyWriter线程服务的流程图可以归纳为如下所示:

然后我们结合LazyWriter和RamDiskReplicaTracker跟踪服务,就可以得到下面一个完整的流程(暂且不考虑RamDiskAsyncLazyPersistService的内部执行逻辑).

RamDiskAsyncLazyPersistService

一个磁盘服务对应一个线程池,并且一个线程池的最大线程数也只有1个.

线程池列表定义如下

class RamDiskAsyncLazyPersistService {
...
  private Map<File, ThreadPoolExecutor> executors
      = new HashMap<File, ThreadPoolExecutor>();
...

这里的File代表的是一个独立的磁盘所在目录,个人认为这里完全可以用String字符串替代.既可以减少存储空间,又直观明了.所以在这里就可以看出是1对1的关系了.

当服务启动的时候,就会有新的磁盘目录加入.

  synchronized void addVolume(File volume) {
    if (executors == null) {
      throw new RuntimeException("AsyncLazyPersistService is already shutdown");
    }
    ThreadPoolExecutor executor = executors.get(volume);
    // 如果当前已存在此磁盘目录对应的线程池,则跑异常
    if (executor != null) {
      throw new RuntimeException("Volume " + volume + " is already existed.");
    }
    // 否则进行添加
    addExecutorForVolume(volume);
  }

进入addExecutorForVolume方法

  private void addExecutorForVolume(final File volume) {
    ...
    // 新建线程池,最大线程执行数为
    ThreadPoolExecutor executor = new ThreadPoolExecutor(
        CORE_THREADS_PER_VOLUME, MAXIMUM_THREADS_PER_VOLUME,
        THREADS_KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,
        new LinkedBlockingQueue<Runnable>(), threadFactory);

    // This can reduce the number of running threads
    executor.allowCoreThreadTimeOut(true);
    // 加入到executors中,以为volume作为key
    executors.put(volume, executor);
  }

还有一个需要注意的是提交执行方法submitLazyPersistTask.

  void submitLazyPersistTask(String bpId, long blockId,
      long genStamp, long creationTime,
      File metaFile, File blockFile,
      FsVolumeReference target) throws IOException {
    if (LOG.isDebugEnabled()) {
      LOG.debug("LazyWriter schedule async task to persist RamDisk block pool id: "
          + bpId + " block id: " + blockId);
    }
    // 获取需要持久化到目标磁盘实例
    FsVolumeImpl volume = (FsVolumeImpl)target.getVolume();
    File lazyPersistDir  = volume.getLazyPersistDir(bpId);
    if (!lazyPersistDir.exists() && !lazyPersistDir.mkdirs()) {
      FsDatasetImpl.LOG.warn("LazyWriter failed to create " + lazyPersistDir);
      throw new IOException("LazyWriter fail to find or create lazy persist dir: "
          + lazyPersistDir.toString());
    }
    // 新建此服务Task
    ReplicaLazyPersistTask lazyPersistTask = new ReplicaLazyPersistTask(
        bpId, blockId, genStamp, creationTime, blockFile, metaFile,
        target, lazyPersistDir);
    // 提交到对应volume的线程池中执行
    execute(volume.getCurrentDir(), lazyPersistTask);
  }

如果在上述执行的过程中发生失败,会调用失败处理的方法,并会重新将此副本块插入到replicateNotPersisted队列等待下一次的持久化.

  public void onFailLazyPersist(String bpId, long blockId) {
    RamDiskReplica block = null;
    block = ramDiskReplicaTracker.getReplica(bpId, blockId);
    if (block != null) {
      LOG.warn("Failed to save replica " + block + ". re-enqueueing it.");
      // 重新插入队列操作
      ramDiskReplicaTracker.reenqueueReplicaNotPersisted(block);
    }
  }

其他的removeVolume等方法实现比较简单,这里不做过多的介绍.下面是RamDiskAsyncLazyPersistService总的结构图:

综合以上3部分的内容阐述,主要描述了LAZT_PERSIST下的FIFO先进先出的队列式内存数据块持久化的顺序,异步持久化服务的内部运行逻辑和LRU算法移除数据副本块来预留内存空间.

LAZY_PERSIST内存存储的使用

首先要使用LAZY_PERSIST内存存储策略,需要有对应的存储介质,内存存储介质对应的类型是RAM_DISK.

所以第一步,需要将机器中已经完成好的RAM disk虚拟内存盘配置到配置项dfs.datanode.data.dir中,其次还要带上,RAM_DISK的标签.如下:

<property>
      <name>dfs.datanode.data.dir</name>
      <value>/grid/0,/grid/1,/grid/2,[RAM_DISK]/mnt/dn-tmpfs</value>
    </property>

注意,这个标签是必须要打上的,否则HDFS都默认的是DISK.

第二步就是设置具体的文件的策略类型了,上文中已经提到过了.

然后附带2个注意事项:

• 确保HDFS异构存储策略没有被关闭,默认是开启的,配置想是dfs.storage.policy.enabled.
• 确保配置dfs.datanode.max.locked.memory是否设置了足够大的内存值,是否已是DataNode能承受的最大内存大小.否则过小会导致内存中的总的可存储的数据块变少,如果超过的话,直接移出.

相关链接

2.百度百科.tmpfs

3.百度百科.RAM disk

4.百度百科.LRU算法