HBase源码分析之MemStore的flush发起时机、判断条件等详情

        前面的几篇文章,我们详细介绍了HBase中HRegion上MemStore的flsuh流程,以及HRegionServer上MemStore的flush处理流程。那么,flush到底是在什么情况下触发的呢?本文我们将详细探究下HBase中MemStore的flush流程的发起时机,看看到底都有哪些操作,或者哪些后台服务进程会触发MemStore的flush。

        首先,在《HBase源码分析之HRegionServer上MemStore的flush处理流程(一)》《HBase源码分析之HRegionServer上MemStore的flush处理流程(二)》两篇文章中,我们了解了HRegionServer上MemStore的flush处理流程,知道了这么一个事实:flush的请求会通过requestFlush()或requestDelayedFlush()方法被添加到MemStoreFlusher的flushQueue队列中,然后由其内部的FlushHandler线程组消费,对需要flush的HRegion进行处理。所以,我们首先可以知道,调用MemStoreFlusher这两个方法的地方,肯定就会是MemStore发起flush的部分时机,另外,或许会存在部分操作或者内部流程直接调用HRegion的flushcache()方法而触发flush。下面,我们开始总结下,都有哪些操作或者内部流程触发MemStore的flush,以及需要flush的判断条件,包括其他一些方面的详情。

        一、通过将RequestFlush请求添加到MemStoreFlusher的flushQueue队列

        (一)单个Put操作

        在HRegion中处理Put操作的put(Put put)方法中,在开始执行操作前,首先会调用checkResources()方法检查资源,这个checkResources()实际上就是检查HRegion的MemStore大小是否超过一定的阈值,如果超过,则会调用requestFlush()方法发起对该HRegion的MemStore进行flush的请求,并抛出RegionTooBusyException异常,阻止该操作继续,后续将要讲的Delete、Append等数据更新操作也是如此,在开始执行操作前都会调用这个checkResources()方法来检查资源。checkResources()方法代码如下:

/*
   * Check if resources to support an update.
   * 检测是否有足够的资源支持一个Put、Append等数据更新操作
   *
   * We throw RegionTooBusyException if above memstore limit
   * and expect client to retry using some kind of backoff
   * 如果超过memstore的限制,我们抛出RegionTooBusyException这个异常,并且期望客户端使用某种补偿进行重试
  */
  private void checkResources()
    throws RegionTooBusyException {
    // If catalog region, do not impose resource constraints or block updates.
	// 如果是Meta Region,不实施资源约束或阻塞更新
    if (this.getRegionInfo().isMetaRegion()) return;

    // 如果Region当前内存大小超过阈值
    // 这个memstoreSize是当前时刻HRegion上MemStore的大小,它是在Put、Append等操作中调用addAndGetGlobalMemstoreSize()方法实时更新的。
    // 而blockingMemStoreSize是HRegion上设定的MemStore的一个阈值,当MemStore的大小超过这个阈值时,将会阻塞数据更新操作
    if (this.memstoreSize.get() > this.blockingMemStoreSize) {
      // 更新阻塞请求计数器
      blockedRequestsCount.increment();
      // 请求刷新Region
      requestFlush();
      
      // 抛出RegionTooBusyException异常
      throw new RegionTooBusyException("Above memstore limit, " +
          "regionName=" + (this.getRegionInfo() == null ? "unknown" :
          this.getRegionInfo().getRegionNameAsString()) +
          ", server=" + (this.getRegionServerServices() == null ? "unknown" :
          this.getRegionServerServices().getServerName()) +
          ", memstoreSize=" + memstoreSize.get() +
          ", blockingMemStoreSize=" + blockingMemStoreSize);
    }
  }
        首先,如果是Meta Region,不实施资源约束或阻塞更新;

        然后,如果Region当前内存大小memstoreSize超过阈值blockingMemStoreSize,则更新阻塞请求计数器,发起刷新MemStore请求,并抛出RegionTooBusyException异常,阻塞数据更新操作。

        我们首先来看下memstoreSize,这个memstoreSize是当前时刻HRegion上MemStore的大小,它是在Put、Append等操作中调用addAndGetGlobalMemstoreSize()方法实时更新的。代码如下:

  /**
   * Increase the size of mem store in this region and the size of global mem
   * store
   * @return the size of memstore in this region
   */
  public long addAndGetGlobalMemstoreSize(long memStoreSize) {
    if (this.rsAccounting != null) {
      rsAccounting.addAndGetGlobalMemstoreSize(memStoreSize);
    }
    return this.memstoreSize.addAndGet(memStoreSize);
  }
        而blockingMemStoreSize是HRegion上设定的MemStore的一个阈值,当MemStore的大小超过这个阈值时,将会阻塞数据更新操作。其定义在HRegion上线被构造时需要调用的一个setHTableSpecificConf()中,部分代码如下:

    // blockingMemStoreSize是HRegion上设定的MemStore的一个阈值,当MemStore的大小超过这个阈值时,将会阻塞数据更新操作
    // 它的计算是由memstoreFlushSize乘以一个比例,这个比例取自参数hbase.hregion.memstore.block.multiplier,
    // 该参数未配置的话,则默认为4
    this.blockingMemStoreSize = this.memstoreFlushSize *
        conf.getLong(HConstants.HREGION_MEMSTORE_BLOCK_MULTIPLIER,
                HConstants.DEFAULT_HREGION_MEMSTORE_BLOCK_MULTIPLIER);
        我们可以知道,它的计算是由memstoreFlushSize乘以一个比例,这个比例取自参数hbase.hregion.memstore.block.multiplier,该参数未配置的话,则默认为4。

        那么memstoreFlushSize是什么呢?memstoreFlushSize为HRegion上设定的一个阈值,当MemStore的大小超过这个阈值时,将会发起flush请求。它的计算首先是由Table决定的,即每个表可以设定自己的memstoreFlushSize,通过关键字MEMSTORE_FLUSHSIZE来设定,如果表中未设定,则取参数hbase.hregion.memstore.flush.size,如果参数再无配置的话,则默认为1024*1024*128L,即128M。代码同样在setHTableSpecificConf()这个方法中,如下:

    if (this.htableDescriptor == null) return;
    long flushSize = this.htableDescriptor.getMemStoreFlushSize();

    if (flushSize <= 0) {
      flushSize = conf.getLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE,
        HTableDescriptor.DEFAULT_MEMSTORE_FLUSH_SIZE);
    }
    
    // memstoreFlushSize为HRegion上设定的一个阈值,当MemStore的大小超过这个阈值时,将会发起flush请求
    // 它的计算首先是由Table决定的,即每个表可以设定自己的memstoreFlushSize,通过关键字MEMSTORE_FLUSHSIZE来设定,
    // 如果未设定,则取参数hbase.hregion.memstore.flush.size,参数未配置的话,则默认为1024*1024*128L,即128M
    this.memstoreFlushSize = flushSize;
        接下来,我们看下requestFlush()是如何发送的,代码如下:
private void requestFlush() {
    
	// 判断HRegion上的rsServices是否为空,rsServices为HRegionServer提供的服务类。
	// HRegion通过持有它,才可以获得能够发起执行诸如flush、compact、split等部分操作的工具对象。
	if (this.rsServices == null) {
      return;
    }
	
	// synchronized同步检查writestate的状态,如果writestate的状态为flushRequested,则直接返回,避免重复请求,
	// 否则将writestate的flushRequested设置为true,并继续发起flush请求
    synchronized (writestate) {
      if (this.writestate.isFlushRequested()) {
        return;
      }
      writestate.flushRequested = true;
    }
    
    // Make request outside of synchronize block; HBASE-818.
    // 通过rsServices获得FlushRequester,继而调用其requestFlush()方法,将HRegion自身传入,发起flush请求。
    // 这个FlushRequester就是HRegionServer上的cacheFlusher,它的requestFlush()就会将flush请求加入到请求队列中,利用内部工作线程去处理
    this.rsServices.getFlushRequester().requestFlush(this);
    if (LOG.isDebugEnabled()) {
      LOG.debug("Flush requested on " + this);
    }
  }
        首先,需要做一些必要的判断和状态设置,如下:

        1、判断HRegion上的rsServices是否为空,rsServices为HRegionServer提供的服务类。HRegion通过持有它,才可以获得能够发起执行诸如flush、compact、split等部分操作的工具对象;

        2、检查writestate的状态,如果writestate的状态为flushRequested,则直接返回,避免重复请求,否则将writestate的flushRequested设置为true,并继续发起flush请求。

        检查通过后,就会通过rsServices获得FlushRequester,继而调用其requestFlush()方法,将HRegion自身传入,发起flush请求。这个FlushRequester就是HRegionServer上的cacheFlusher,它的requestFlush()就会将flush请求加入到请求队列中,利用内部工作线程去处理。getFlushRequester()方法代码在实现了RegionServerServices()接口的HRegionServer中,代码如下:

  /** @return reference to FlushRequester */
  @Override
  public FlushRequester getFlushRequester() {
    return this.cacheFlusher;
  }
        继续回到put()方法,继而依次调用doBatchMutate()、batchMutate()、batchMutate()方法,执行Put操作。而batchMutate()方法的定义,是针对批量的Put、Delete等操作而专门设计的一个方法,只不过单个的Put等操作传入的是一个只包含一个操作的数组。batchMutate()方法会在批量操作未全部完成前一直循环,每次循环时,都会调用checkResources()检测MemStore,并调用doMiniBatchMutation()方法完成操作并同步更新HRegion的MemStore大小,获取其值为newSize,最后通过isFlushSize()方法判断是否需要发起一个flush请求来决定是否调用requestFlush()方法。代码如下:

**
   * Perform a batch of mutations.
   * It supports only Put and Delete mutations and will ignore other types passed.
   * 
   * 完成一批变化操作。
   * 它仅支持Put和Delete操作,将忽略其他类型的操作。
   * 
   * @param batchOp contains the list of mutations
   * @return an array of OperationStatus which internally contains the
   *         OperationStatusCode and the exceptionMessage if any.
   * @throws IOException
   */
  OperationStatus[] batchMutate(BatchOperationInProgress<?> batchOp) throws IOException {
    boolean initialized = false;
    // 操作是否为回放,回放的话为REPLAY_BATCH_MUTATE,否则为BATCH_MUTATE
    Operation op = batchOp.isInReplay() ? Operation.REPLAY_BATCH_MUTATE : Operation.BATCH_MUTATE;
    // 开始Region上的操作
    startRegionOperation(op);
    try {
    	
      // 操作未全部完成前一直循环
      while (!batchOp.isDone()) {
        // 如果是回放操作,则检测ReadOnly
    	if (!batchOp.isInReplay()) {
          checkReadOnly();
        }
    	
    	// 检测相关资源
        checkResources();

        // 未初始化的话要先初始化
        if (!initialized) {
          // 更新写请求计数器
          this.writeRequestsCount.add(batchOp.operations.length);
          // 如果不是日志回放,执行pre钩子方法
          if (!batchOp.isInReplay()) {
            doPreMutationHook(batchOp);
          }
          initialized = true;
        }
        // 执行操作,并返回增加的内存大小
        long addedSize = doMiniBatchMutation(batchOp);
        
        // 以原子操作的方式增加Region上的MemStore内存大小
        long newSize = this.addAndGetGlobalMemstoreSize(addedSize);
        
        // 内存超过阈值时,请求flush
        if (isFlushSize(newSize)) {
          requestFlush();
        }
      }
    } finally {
      // 关闭Region上的操作
      closeRegionOperation(op);
    }
    
    // 返回批操作的状态
    return batchOp.retCodeDetails;
  }
        我们来看下isFlushSize()方法,很简单,它就是通过判断当前MemStore大小newSize是否超过memstoreFlushSize来决定。代码如下:

/*
   * @param size
   * @return True if size is over the flush threshold
   */
  private boolean isFlushSize(final long size) {
    return size > this.memstoreFlushSize;
  }
        至此,单个Put操作讲完了。

        (二)单个Delete操作

        单个Delete操作与单个Put操作一样,代码如下:

//////////////////////////////////////////////////////////////////////////////
  // set() methods for client use.
  //////////////////////////////////////////////////////////////////////////////
  /**
   * @param delete delete object
   * @throws IOException read exceptions
   */
  public void delete(Delete delete)
  throws IOException {
    checkReadOnly();
    checkResources();
    startRegionOperation(Operation.DELETE);
    try {
      delete.getRow();
      // All edits for the given row (across all column families) must happen atomically.
      doBatchMutate(delete);
    } finally {
      closeRegionOperation(Operation.DELETE);
    }
  }
        也是先调用checkResources()检查MemStore,再调用doBatchMutate()进行处理,同单个Put操作是一样。读者可自行分析。

        (三)checkAndMutate/checkAndRowMutate操作

        checkAndMutate()/checkAndRowMutate()方法中,也是先调用checkResources()检查MemStore,再调用doBatchMutate()进行处理,同单个Put操作是一样。checkAndMutate/checkAndRowMutate操作的特点就是保证修改数据的原子性,也是属于数据更新的操作。

        (四)单个Append操作

        在HBase中的append()方法中,同样是先调用checkResources()方法检测HRegion的MemStore,然后在处理完append操作后,调用addAndGetGlobalMemstoreSize()方法更新并获取最新的MemStore大小size,继而调用isFlushSize()方法判断是否需要调用requestFlush()方法发起flush请求。处理模式也是同上述的单个Put、单个Delete操作大致相同。

        (五)单个Increment操作

        同单个Append操作,在HBase中的increment()方法中,对于MemStore的检查及判断是否需要flush等完全一致,不再赘述。

        (六)批量操作

        既然单个操作可能会引起flush,那么批量操作更不用说了。批量操作是通过batchMutate()方法实现的,上面已经分析过这个方法了,这里不再赘述。

        综上所述,凡是涉及到数据更新的操作,比如Put、Delete、Append、increment等操作,均是先检查MemStore,如果其高于某个阈值,将会发送flush请求,并抛出异常,阻塞数据更新操作。另外,在操作执行完毕后,也会根据MemStore增长情况,判断是否达到了该触发flush的条件,如果条件满足,则会发送flush请求。

        二、通过直接调用HRegion的flushcache()方法

        (一)外部触发(命令行等)

        这种场景的是通过RegionServer上RSRpcServices的flushRegion()方法发起的。flushRegion()的代码如下:

  /**
   * Flush a region on the region server.
   * 刷新RegionServer上的 一个Region
   *
   * @param controller the RPC controller
   * @param request the request
   * @throws ServiceException
   */
  @Override
  @QosPriority(priority=HConstants.ADMIN_QOS)
  public FlushRegionResponse flushRegion(final RpcController controller,
      final FlushRegionRequest request) throws ServiceException {
    try {
      // 检测RegionServer的状态
      checkOpen();
      
      // 请求计数器加1
      requestCount.increment();
      
      // 获取需要flush的Region
      HRegion region = getRegion(request.getRegion());
      LOG.info("Flushing " + region.getRegionNameAsString());
      boolean shouldFlush = true;
      
      // 判断是否需要flush,判断的依据是region上次flush时间小于请求中getIfOlderThanTs()方法的返回值
      if (request.hasIfOlderThanTs()) {
        shouldFlush = region.getLastFlushTime() < request.getIfOlderThanTs();
      }
      
      FlushRegionResponse.Builder builder = FlushRegionResponse.newBuilder();
      
      // 如果应该刷新
      if (shouldFlush) {
    	// 开始时间
        long startTime = EnvironmentEdgeManager.currentTime();
        // 调用Region的flushcache()方法刷新
        HRegion.FlushResult flushResult = region.flushcache();
        if (flushResult.isFlushSucceeded()) {// 刷新成功的话,记录刷新耗时
          long endTime = EnvironmentEdgeManager.currentTime();
          regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);
        }
        boolean result = flushResult.isCompactionNeeded();
        if (result) {// 需要合并的话,通过RegionServer的compactSplitThread发起合并请求
          regionServer.compactSplitThread.requestSystemCompaction(region,
            "Compaction through user triggered flush");
        }
        builder.setFlushed(result);
      }
      builder.setLastFlushTime(region.getLastFlushTime());
      return builder.build();
    } catch (DroppedSnapshotException ex) {
      // Cache flush can fail in a few places. If it fails in a critical
      // section, we get a DroppedSnapshotException and a replay of wal
      // is required. Currently the only way to do this is a restart of
      // the server.
      regionServer.abort("Replay of WAL required. Forcing server shutdown", ex);
      throw new ServiceException(ex);
    } catch (IOException ie) {
      throw new ServiceException(ie);
    }
  }
        flushRegion()方法内部在判断应该进行flush后,会调用HRegion的flushcache()方法对其MemStore进行flush处理,代码很简单,这里就不多说了。
        (二)Region合并(不是compact,而是两个Region的merge)

        在RSRpcServices中存在Region合并时调用的mergeRegions()方法,在其内部会先后调用regionA和regionB的flushcache()方法去flush每个Region上的MemStore,然后再执行Region合并。关键代码如下:

      long startTime = EnvironmentEdgeManager.currentTime();
      
      // 刷新regionA的MemStore
      HRegion.FlushResult flushResult = regionA.flushcache();
      if (flushResult.isFlushSucceeded()) {
        long endTime = EnvironmentEdgeManager.currentTime();
        regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);
      }
      startTime = EnvironmentEdgeManager.currentTime();
      
      // 刷新regionB的MemStore
      flushResult = regionB.flushcache();
      if (flushResult.isFlushSucceeded()) {
        long endTime = EnvironmentEdgeManager.currentTime();
        regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);
      }
      regionServer.compactSplitThread.requestRegionsMerge(regionA, regionB, forcible,
          masterSystemTime);

        (三)Region分裂

         在RSRpcServices中存在Region分裂时调用的splitRegion()方法,也是先调用flushcache()将Region上的memstore刷新。代码如下:

/**
   * Split a region on the region server.
   * 分裂RegionServer上的一个Region
   *
   * @param controller the RPC controller
   * @param request the request
   * @throws ServiceException
   */
  @Override
  @QosPriority(priority=HConstants.ADMIN_QOS)
  public SplitRegionResponse splitRegion(final RpcController controller,
      final SplitRegionRequest request) throws ServiceException {
    try {
      
      // 检测RegionServer状态
      checkOpen();
      // 请求计数器加1
      requestCount.increment();
      // 获取对应的Region
      HRegion region = getRegion(request.getRegion());
      // Region上开启一个SPLIT_REGION操作
      region.startRegionOperation(Operation.SPLIT_REGION);
      LOG.info("Splitting " + region.getRegionNameAsString());
      
      // Split开始时间
      long startTime = EnvironmentEdgeManager.currentTime();
      
      // 先调用flushcache()将Region上的memstore刷新
      HRegion.FlushResult flushResult = region.flushcache();
      if (flushResult.isFlushSucceeded()) {
    	// flush成功的话,RegionServer记录flush耗时
        long endTime = EnvironmentEdgeManager.currentTime();
        regionServer.metricsRegionServer.updateFlushTime(endTime - startTime);
      }
      byte[] splitPoint = null;
      // 如果请求中有切分点,则取请求中的切分点
      if (request.hasSplitPoint()) {
        splitPoint = request.getSplitPoint().toByteArray();
      }
      
      // 设置标志位splitRequest为tue,并设置明确的分裂点explicitSplitPoint
      region.forceSplit(splitPoint);
      
      // 调用RegionServer上compactSplitThread的requestSplit()方法,请求分裂region
      // 调用前,先调用Region的checkSplit()方法,确定是否可以切分,并确定切分点
      regionServer.compactSplitThread.requestSplit(region, region.checkSplit());
      return SplitRegionResponse.newBuilder().build();
    } catch (DroppedSnapshotException ex) {
      regionServer.abort("Replay of WAL required. Forcing server shutdown", ex);
      throw new ServiceException(ex);
    } catch (IOException ie) {
      throw new ServiceException(ie);
    }
  }

        (四)利用Bulk加载HFile

        Bulk是HBase直接加载HFile存储数据的一种高速、实用的手段或工具。在HRegion中的bulkLoadHFiles()方法中,也会调用flushcache()方法刷新HRegion上的MemStore内存。相关代码如下:

if (assignSeqId) {
        FlushResult fs = this.flushcache();
        if (fs.isFlushSucceeded()) {
          seqId = fs.flushSequenceId;
        } else if (fs.result == FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY) {
          seqId = fs.flushSequenceId;
        } else {
          throw new IOException("Could not bulk load with an assigned sequential ID because the " +
              "flush didn't run. Reason for not flushing: " + fs.failureReason);
        }
      }

        (五)做Table的快照

        在做表的快照时,会将对应Table中涉及到的Region的MemStore进行flush,做这项工作的是FlushSnapshotSubprocedure类中的RegionSnapshotTask,它在call()方法中,当snapshotSkipFlush为false时,会调用HRegion的flushcache()方法,对MemStore进行flush。相关代码如下:

 if (snapshotSkipFlush) {
        /*
         * This is to take an online-snapshot without force a coordinated flush to prevent pause
         * The snapshot type is defined inside the snapshot description. FlushSnapshotSubprocedure
         * should be renamed to distributedSnapshotSubprocedure, and the flush() behavior can be
         * turned on/off based on the flush type.
         * To minimized the code change, class name is not changed.
         */
          LOG.debug("take snapshot without flush memstore first");
        } else {
          LOG.debug("Flush Snapshotting region " + region.toString() + " started...");
          region.flushcache();
        }

        综上所述,凡是涉及到Region个数改变(比如两个变一个的合并、一个变两个的分裂、一个拷贝一个的快照等),大都需要先flush掉其MemStore。但是有个疑问,为什么非要做flush呢?将内存中的数据也合并、分裂或者拷贝,不是更快吗?难道是因为其不稳定?留待以后再深入分析吧!

        好了,MemStore的flush发起时机、判断条件等详情基本上就这些了,如果有遗漏的,或者讲解不深入的,以后再慢慢深入吧!




上一篇:GraphX 在图数据库 Nebula Graph 的图计算实践


下一篇:HBase源码分析之HRegionServer上compact流程分析