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

简介:         在《HBase源码分析之MemStore的flush发起时机、判断条件等详情》一文中,我们详细介绍了MemStore flush的发起时机、判断条件等详情,主要是两类操作,一是会引起MemStore数据大小变化的Put、Delete、Append、Increment等操作,二是会引起HRegion变化的诸如Regin的分裂、合并以及做快照时的复制拷贝等,同样会触发MemStore的flush流程。

        在《HBase源码分析之MemStore的flush发起时机、判断条件等详情》一文中,我们详细介绍了MemStore flush的发起时机、判断条件等详情,主要是两类操作,一是会引起MemStore数据大小变化的Put、Delete、Append、Increment等操作,二是会引起HRegion变化的诸如Regin的分裂、合并以及做快照时的复制拷贝等,同样会触发MemStore的flush流程。同时,在《HBase源码分析之compact请求发起时机、判断条件等详情(一)》一文中,我们讲到了针对compact,在HRegionServer内部存在一个工作线程compactionChecker,它会周期性的工作,以检查是否达到能够发起compact请求的条件。那么,回过头来,我们再看MemStore flush,它是不是也存在一个后台工作线程,能够周期性的工作,以检查是否达到能够发起flush请求的条件呢?本文,我们就之前《HBase源码分析之MemStore的flush发起时机、判断条件等详情》一文,做一个关于MemStore flush后台检查线程等内容的补充。

        在HRegionServer中,有一个和合并检查线程compactionChecker一样的Chore--periodicFlusher,它也是类似于compactionChecker的后台工作线程,它负责周期性的检查MemStore,查看是否达到发起MemStore flush的条件。其定义如下:

  /*
   * Check for flushes
   * 检查刷新请求
   */
  Chore periodicFlusher;
        它也是一个继承自Chore的工作线程,关于Chore的介绍,在 《HBase源码分析之compact请求发起时机、判断条件等详情(一)》一文中我已经讲过了,这里不再做介绍。而periodicFlusher的初始化,自然同compactionChecker一样,也是在HRegionServer的initializeThreads()方法中完成的,代码如下:

    this.periodicFlusher = new PeriodicMemstoreFlusher(this.threadWakeFrequency, this);
        很简单,构造一个PeriodicMemstoreFlusher对象,而且其工作频率也是通过HRegionServer的threadWakeFrequency决定的。那么这个periodicFlusher到底是什么样的实现类,其工作原理是什么样子的呢?莫慌,让我为大家一一道来。

        首先看下PeriodicMemstoreFlusher的定义、成员变量与构造方法,代码如下:

  static class PeriodicMemstoreFlusher extends Chore {
    final HRegionServer server;
    final static int RANGE_OF_DELAY = 20000; //millisec
    final static int MIN_DELAY_TIME = 3000; //millisec
    public PeriodicMemstoreFlusher(int cacheFlushInterval, final HRegionServer server) {
      // cacheFlushInterval为flush的时间间隔
      super(server.getServerName() + "-MemstoreFlusherChore", cacheFlushInterval, server);
      this.server = server;
    }
  }
        通过成员变量和构造方法,我们可以看到,比较重要的就是线程中HRegionServer的实例server以及线程工作频率。另外它还提供了两个定值MIN_DELAY_TIME与RANGE_OF_DELAY,有什么用呢。继续看它的chore()方法:

    @Override
    protected void chore() {
      // 循环HRegionSever上的onlineRegions
      for (HRegion r : this.server.onlineRegions.values()) {
        
    	// HRegion为null的话直接跳过
    	if (r == null)
          continue;
        
    	// 调用HRegion上的shouldFlush()方法,判断是否可以进行flush
        if (r.shouldFlush()) {
          // 获取RegionServer上的MemStoreFlusher类型的memstore内存刷新管理对象
          FlushRequester requester = server.getFlushRequester();
          if (requester != null) {
        	// 随机延迟时间:20s内的一个随机时间+3s的基础时间
            long randomDelay = RandomUtils.nextInt(RANGE_OF_DELAY) + MIN_DELAY_TIME;
            LOG.info(getName() + " requesting flush for region " + r.getRegionNameAsString() +
                " after a delay of " + randomDelay);
            //Throttle the flushes by putting a delay. If we don't throttle, and there
            //is a balanced write-load on the regions in a table, we might end up
            //overwhelming the filesystem with too many flushes at once.
            // 通过设置一个延迟时间控制flush,防止Region上多个flush同一时间并发进行
            requester.requestDelayedFlush(r, randomDelay);
          }
        }
      }
    }
  }
        通过chore()方法我们知道,periodicFlusher线程周期性的对HRegionServer上所有在线Region进行检测,调用其shouldFlush()方法进行检测,如果该Region需要flush memstore,获取RegionServer上的MemStoreFlusher类型的memstore内存刷新管理对象,发起flush请求。

        需要注意的是,该flush请求携带一个固定加随机的延迟时间,其算法为:

long randomDelay = RandomUtils.nextInt(RANGE_OF_DELAY) + MIN_DELAY_TIME;
        MIN_DELAY_TIME就是我们上面提到的固定值3秒,然后再加上一个20s内的一个随机数。为什么要这么做呢?试想下,如果立即提交一个flush请求,或者在3秒后立即提交一个flush请求,是不是很容易就产生一个风暴,引起系统性能瓶颈呢?

        关于如何提交一个flush请求,前面的文章已经介绍过了,不再赘述。这里我们介绍下HRegion的shouldFlush()方法,代码如下:

  /**
   * Should the memstore be flushed now
   * memstore现在是否应该被flush
   */
  boolean shouldFlush() {
    // This is a rough measure.
	// 这里是一个粗略的测量
	// 上次flush之后,sequenceId的增长超过flushPerChanges,即发起一次flush
	// 次数限制通过参数hbase.regionserver.flush.per.changes配置,默认为30000000(3千万)
	// 也就是该Region上数据的改动次数,无论增、改、删等,超过一定的次数,即发起一次flush
	// 意味着会兼顾HRegion上的写请求及时flush到磁盘上
    if (this.lastFlushSeqId > 0
          && (this.lastFlushSeqId + this.flushPerChanges < this.sequenceId.get())) {
      return true;
    }
    
    // hbase.regionserver.optionalcacheflushinterval参数小于等于0,不会触发flush
    //
    if (flushCheckInterval <= 0) { //disabled
      return false;
    }
    long now = EnvironmentEdgeManager.currentTime();
    //if we flushed in the recent past, we don't need to do again now
    // 时间间隔未超过hbase.regionserver.optionalcacheflushinterval配置的时间间隔
    // 默认为3600000ms,即1小时
    if ((now - getLastFlushTime() < flushCheckInterval)) {
      return false;
    }
    
    //since we didn't flush in the recent past, flush now if certain conditions
    //are met. Return true on first such memstore hit.
    // 检测每个列簇,当其中一个列簇超过flushCheckInterval没有flush时,发起flush
    for (Store s : this.getStores().values()) {
      if (s.timeOfOldestEdit() < now - flushCheckInterval) {
        // we have an old enough edit in the memstore, flush
        return true;
      }
    }
    return false;
  }
        判断的逻辑比较清晰,概括如下:

        1、首先,上次flush之后,sequenceId的增长超过flushPerChanges,即发起一次flush:

              次数限制flushPerChanges是通过参数hbase.regionserver.flush.per.changes配置,默认为30000000(3千万),这个sequenceId的增长该Region上数据的改动次数,无论增、删、改或者append、increment等,它是对HRegion数据变动的一个考虑,即便是MemStore不大,数据变动的频繁了,也需要进行flush,以降低宕机后拆分日志的工作量;

        2、再看参数hbase.regionserver.optionalcacheflushinterval:

              参数小于等于0,不会触发flush,时间间隔未超过参数l配置的时间间隔的话,也不会触发flush。这个参数默认为3600000ms,即1小时;

        3、当超过参数配置的时间间隔,再检测每个列簇,当其中一个列簇超过flushCheckInterval没有flush时,发起flush,也就是说它有足够久的数据没有被flush。

        以上就是HRegionServer内部PeriodicMemstoreFlusher工作线程periodicFlusher的全部内容。同时,在上面针对每个HRegion的循环,以及后面针对每个HStore的判断,我们可以发现,flush还是以Region为最小单位进行的。即便是某个列簇下MemStore过大或者过旧,另外一个MemStore还比较小或者比较新的话,它还是跟着那个过大或者过旧的列簇一起flush,这也是HBase饱受诟病的列簇不能过多的原因之一。在HBase1.1.2版本中,有对于MemStore flush的改进,改成了以HStore,即列簇为单位进行。此乃后话,我们以后再做分析。







相关实践学习
云数据库HBase版使用教程
&nbsp; 相关的阿里云产品:云数据库 HBase 版 面向大数据领域的一站式NoSQL服务,100%兼容开源HBase并深度扩展,支持海量数据下的实时存储、高并发吞吐、轻SQL分析、全文检索、时序时空查询等能力,是风控、推荐、广告、物联网、车联网、Feeds流、数据大屏等场景首选数据库,是为淘宝、支付宝、菜鸟等众多阿里核心业务提供关键支撑的数据库。 了解产品详情:&nbsp;https://cn.aliyun.com/product/hbase &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
相关文章
|
3月前
|
SQL 关系型数据库 分布式数据库
Flink报错问题之用flush方法写入hbase报错如何解决
Apache Flink是由Apache软件基金会开发的开源流处理框架,其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。
|
Java 测试技术 分布式数据库
从数据结构比较HBase的3种memstore实现方案
HBase的memstore目前存在3种实现:DefaultMemstore、CompactingMemstore、CCSMapMemStore,本文尝试从数据结构的角度对其进行比较。
1556 0
|
分布式数据库 Hbase Java
hbase region split源码分析
hbase region split : split执行调用流程: 1.HbaseAdmin发起split:### 2.RSRpcServices实现类执行split(Implements the regionserver RPC services.)### 3.CompactSplitThread类与SplitRequest类用来执行region切割:### 4.splitRequest执行doSplitting操作### 4.1初始化两个子region### 4.2执行切割#### 4.2.1:(创建子region。
1759 0
|
分布式数据库 Hbase
hbase snapshot源码分析
snapshot操作在硬盘上形式: /hbase/.snapshots /.tmp &lt;---- working directory /[snapshot name] &lt;---...
1340 0
|
分布式数据库 Hbase C++
HBase源码分析之HRegionServer上MemStore的flush处理流程(二)
        继上篇文章《HBase源码分析之HRegionServer上MemStore的flush处理流程(一)》遗留的问题之后,本文我们接着研究HRegionServer上MemStore的flush处理流程,重点讲述下如何选择一个HRegion进行flush以缓解MemStore压力,还有HRegion的flush是如何发起的。
983 0
|
Java 分布式数据库 Go
HBase的scan源码分析客户端部分之整体流程(一)
        scan的调用代码示例如下: // 创建HBase配置config Configuration config = HBaseConfiguration.create(); config.
1807 0
|
大数据 分布式数据库 Hbase
写在HBase源码分析前面的话
       本版块全部为HBase1.0.2相关源码分析文章,系个人研究源码原创写成,除对部分引用标示外,其余均为原创,或翻译源码注释。        该系列文章为与网友交流HBase学习,不做任何其他商业用途~~O(∩_∩)O哈哈~         由于水平有限,文章基本上是边读源码,边翻译注释,边分析源码写成,没有较强的前后逻辑性,我会在写完全部文章后再回头整理。
958 0
|
5月前
|
Java Shell 分布式数据库
【大数据技术Hadoop+Spark】HBase数据模型、Shell操作、Java API示例程序讲解(附源码 超详细)
【大数据技术Hadoop+Spark】HBase数据模型、Shell操作、Java API示例程序讲解(附源码 超详细)
87 0
|
9月前
|
SQL 分布式计算 Hadoop
Hadoop集群hbase的安装
Hadoop集群hbase的安装
147 0
|
19天前
|
分布式计算 监控 Hadoop
Ganglia监控Hadoop与HBase集群
Ganglia监控Hadoop与HBase集群