hbase写数据过程

博文说明：1、研究版本hbase0.94.12；2、贴出的源代码可能会有删减，只保留关键的代码

 

从client和server两个方面探讨hbase的写数据过程。

一、client端

 

1、写数据API

    写数据主要是HTable的单条写和批量写两个API，源码如下：

//单条写API

public void put(final Put put) throws IOException {

    doPut(put);

    if (autoFlush) {

      flushCommits();

    }

  }

//批量写API

public void put(final List<Put> puts) throws IOException {

    for (Put put : puts) {

      doPut(put);

    }

    if (autoFlush) {

      flushCommits();

    }

  }

//具体的put实现

private void doPut(Put put) throws IOException{

    validatePut(put);

    writeBuffer.add(put);

    currentWriteBufferSize += put.heapSize();

    if (currentWriteBufferSize > writeBufferSize) {

      flushCommits();

    }

  }

public void close() throws IOException {

    if (this.closed) {

      return;

    }

    flushCommits();

    ….

  }

    通过两个put API可以看出如果autoFlush为false，则无论是否是批量写效果均是相同，均是等待写入的数据超过配置的writeBufferSize（通过hbase.client.write.buffer配置，默认为2M）时才提交写数据请求，如果最后的写入数据没有超过2M，则在调用close方法时会进行最后的提交，当然，如果使用批量的put方法时，自己控制flushCommits则效果不同，比如每隔1000条进行一次提交，如果1000条数据的总大小超过了2M，则实际上会发生多次提交，导致最终的提交次数多过只由writeBufferSize控制的提交次数，因此在实际的项目中，如果对写性能的要求比对数据的实时可查询和不可丢失的要求更高则可以设置autoFlush为false并采用单条写的put(final Put put)API，这样即可以简化写操作数据的程序代码，写入效率也更优，需要注意的是如果对数据的实时可查询和不可丢失有较高的要求则应该设置autoFlush为true并采用单条写的API，这样可以确保写一条即提交一条。

 

2、关于多线程写

通过HConnection的getTable方法获取的HTable对象进行put操作时默认就是多线程的操作，线程数与put涉及的region数有关，虽然是hbase内部是多线程，但是在进行写操作时还是需要自己写多线程进行处理，这样可以大大的提高写速度，相关源码如下，相关源码如下：

public void flushCommits() throws IOException {

    try {

      Object[] results = new Object[writeBuffer.size()];

      try {

        this.connection.processBatch(writeBuffer, tableName, pool, results);

      } catch (InterruptedException e) {

        throw new IOException(e);

      } finally {

        …

    } finally {

      …

    }

  }

 

public void processBatch(List<? extends Row> list, final byte[] tableName, ExecutorService pool,

        Object[] results) throws IOException, InterruptedException {

      …

      processBatchCallback(list, tableName, pool, results, null);

    }

 

public <R> void processBatchCallback(List<? extends Row> list, byte[] tableName, ExecutorService pool,

        Object[] results, Batch.Callback<R> callback) throws IOException, InterruptedException {

      ….

      HRegionLocation [] lastServers = new HRegionLocation[results.length];

      for (int tries = 0; tries < numRetries && retry; ++tries) {

…

        // step 1: break up into regionserver-sized chunks and build the data structs

        Map<HRegionLocation, MultiAction<R>> actionsByServer =

          new HashMap<HRegionLocation, MultiAction<R>>();

for (int i = 0; i < workingList.size(); i++) {

          Row row = workingList.get(i);

          if (row != null) {

            HRegionLocation loc = locateRegion(tableName, row.getRow());

            byte[] regionName = loc.getRegionInfo().getRegionName();

            MultiAction<R> actions = actionsByServer.get(loc);

            if (actions == null) {

              actions = new MultiAction<R>();

              actionsByServer.put(loc, actions);  //每一个region对应一个MultiAction对象，每个MultiAction对象持有该region所有的put Action

            }

            Action<R> action = new Action<R>(row, i);

            lastServers[i] = loc;

            actions.add(regionName, action);

          }

        }

 

        // step 2: make the requests，每个region开启一个线程

        Map<HRegionLocation, Future<MultiResponse>> futures =

            new HashMap<HRegionLocation, Future<MultiResponse>>(actionsByServer.size());

        for (Entry<HRegionLocation, MultiAction<R>> e: actionsByServer.entrySet()) {

          futures.put(e.getKey(), pool.submit(createCallable(e.getKey(), e.getValue(), tableName)));

        }

 

        // step 3: collect the failures and successes and prepare for retry

        …

        // step 4: identify failures and prep for a retry (if applicable).

        …

}

…

}

 

3、在写入数据前，需要先定位具体的数据应该写入的region，核心方法：

//从缓存中定位region，通过NavigableMap实现，如果没有缓存则需查询.META.表

HRegionLocation getCachedLocation(final byte [] tableName,

        final byte [] row) {

      SoftValueSortedMap<byte [], HRegionLocation> tableLocations =

        getTableLocations(tableName);

      …

      //找到小于rowKey并且最接近rowKey的startKey对应的region，通过NavigableMap实现

      possibleRegion = tableLocations.lowerValueByKey(row);

      if (possibleRegion == null) {

        return null;

      }

      //表的最末一个region的endKey是空字符串，如果不是最末一个region，则只有当rowKey小于endKey才返回region。

      byte[] endKey = possibleRegion.getRegionInfo().getEndKey();

      if (Bytes.equals(endKey, HConstants.EMPTY_END_ROW) ||

          KeyValue.getRowComparator(tableName).compareRows(

              endKey, 0, endKey.length, row, 0, row.length) > 0) {

        return possibleRegion;

      }

      return null;

}

 

 二、服务端

    服务端写数据的主要过程是：写WAL日志（如果没有关闭写WAL日志）-》写memstore-》触发flush memstore（如果memstore大小超过hbase.hregion.memstore.flush.size的设置值），在flush memstore过程中可能会触发compact和split操作，在以下内容会对写put方法、flush memstore、compact和split进行讲解。

 

1、HTableInterface接口操作hbase数据的API对应的服务端是由HRegionServer类实现，源代码如下：

 

//单条put

public void put(final byte[] regionName, final Put put) throws IOException {

    HRegion region = getRegion(regionName);

if (!region.getRegionInfo().isMetaTable()) {

//检查HRegionServer的memstore总内存占用量是否已经超过了hbase.regionserver.global.memstore.upperLimit（默认值是0.4）或者hbase.regionserver.global.memstore.lowerLimit（默认值是0.35）的限制，如果超过了则会在flush队列中添加一个任务，其中如果是超过了upper的限制则会阻塞所有的写memstore的操作，直到内存降至upper限制以下。

       this.cacheFlusher.reclaimMemStoreMemory();

    }

boolean writeToWAL = put.getWriteToWAL();

//region会调用Store的add()方法把数据保存到相关Store的memstore中

//region在保存完数据后，会检查是否需要flush memstore，如果需要则发出flush请求，由HRegionServer的flush守护线程异步执行。

    region.put(put, getLockFromId(put.getLockId()), writeToWAL);

  }

//批量put

public int put(final byte[] regionName, final List<Put> puts) throws IOException {

    region = getRegion(regionName);

    if (!region.getRegionInfo().isMetaTable()) {

       this.cacheFlusher.reclaimMemStoreMemory();

    }

   OperationStatus codes[] = region.batchMutate(putsWithLocks);

    for (i = 0; i < codes.length; i++) {

        if (codes[i].getOperationStatusCode() != OperationStatusCode.SUCCESS) {

          return i;

        }

     }

     return -1;

  }

 

2、Flush Memstore

    memstore的flush过程由类MemStoreFlusher控制，该类是Runnable的实现类，在HRegionServer启动时会启动一个MemStoreFlusher的守护线程，每隔10s从flushQueue中获取flush任务进行刷新，如果需要flush memstore时，只需调用MemStoreFlusher的requestFlush或者requestDelayedFlush方法把flush请求加入到flush队列中即可，具体的flush是异步执行的。

    memstore的大小有两个控制级别：

    1）Region级

    a、hbase.hregion.memstore.flush.size：默认值128M，超过将被flush到磁盘

    b、hbase.hregion.memstore.block.multiplier：默认值2，如果memstore的内存大小已经超过了hbase.hregion.memstore.flush.size的2倍，则会阻塞该region的写操作，直到内存大小降至该值以下

  2）RegionServer级

   a、hbase.regionserver.global.memstore.lowerLimit：默认值0.35，HRegionServer的所有memstore占用内存在HRegionServer总内存中占的lower比例，当达到该值，则会从整个RS中找出最需要flush的region进行flush

   b、hbase.regionserver.global.memstore.upperLimit：默认值0.4，HRegionServer的所有memstore占用内存在总内存中的upper比例，当达到该值，则会从整个RS中找出最需要flush的region进行flush，直到总内存比例降至该数限制以下，并且在降至限制比例以下前将阻塞所有的写memstore的操作

  在对整个HRegionServer进行flush操作时，并不会刷新所有的region，而是每次均会根据region的memstore大小、storeFile数量等因素找出最需要flush的region进行flush，flush完成后再进行内存总比例的判断，如果还未降至lower限制以下则会再寻找新的region进行flush。

  在flush region时会flush该region下所有的store，虽然可能某些store的memstore内容很少。

  在flush memstore时会产生updatesLock（HRegion类的一个属性，采用jdk的ReentrantReadWriteLock实现）的排它锁write lock，当获取完memstore的快照后释放updatesLock的write lock，在释放之前，所有的需要获取updatesLock的write、read lock的操作均会被阻塞，该影响是整个HRegion范围，因此如果表的HRegion数量过少，或者数据写入时热点在一个region时会导致该region不断flush memstore，由于该过程会产生write排他锁（虽然进行memstore快照的时间会很快），因此会影响region 的整体写能力。

 

3、Compact操作

  hbase有两种compact：minor和major，minor通常会把若干个小的storeFile合并成一个大的storeFile，minor不会删除标示为删除的数据和过期的数据，major则会删除这些数据，major合并之后，一个store只有一个storeFile文件，这个过程对store的所有数据进行重写，有较大的资源开销，major 合并默认1天执行一次，可以通过hbase.hregion.majorcompaction配置执行周期，通常是把该值设置为0进行关闭，采用手工执行，这样可以避免当集群繁忙时执行整个集群的major合并，major合并是必须执行的操作，因为删除标示为删除和过期的数据操作是在该合并过程中进行的。

    compact合并的级别

    1）、整个hbase集群

在HRegionServer启动时会开启一个守护线程定时扫描集群下的所有在线的region下的storeFile文件，对所有符合Store.needsCompaction()或Store.isMajorCompaction()的store进行合并操作，默认扫描周期是10000秒（大概2.7小时），即大概每隔10000秒进行一次全局的compact，应该尽量减少storefile的文件数，避免每次全局compact时真实发生compact的数量，减少整个集群的负载，可以关闭任何的compact，半夜通过脚本触发：

//threadWakeFrequency默认值是10*1000，multiplier默认值是1000，单位:毫秒

this.compactionChecker = new CompactionChecker(this, this.threadWakeFrequency * multiplier, this);

 

//chore是CompactionChecker定时执行的方法，定时进行minor和major的compcat合并，如果hbase.hregion.majorcompaction配置为0则不执行major合并，minor升级为major除外。

protected void chore() {

      for (HRegion r : this.instance.onlineRegions.values()) {

        if (r == null)

          continue;

        for (Store s : r.getStores().values()) {

          try {

            if (s.needsCompaction()) {

              //如果整个store下的storeFile文件均需要合并，则会自动升级到major合并

              this.instance.compactSplitThread.requestCompaction(r, s, getName()

                  + " requests compaction", null);

            } else if (s.isMajorCompaction()) {

              if (majorCompactPriority == DEFAULT_PRIORITY

                  || majorCompactPriority > r.getCompactPriority()) {

                this.instance.compactSplitThread.requestCompaction(r, s, getName()

                    + " requests major compaction; use default priority", null);

              } else {

                this.instance.compactSplitThread.requestCompaction(r, s, getName()

                    + " requests major compaction; use configured priority",

                  this.majorCompactPriority, null);

              }

            }

          } catch (IOException e) {

            LOG.warn("Failed major compaction check on " + r, e);

          }

        }

      }

}

 

//store内除去正在执行compact的storeFile后剩余的storeFile数如果大于配置的最小可合并数，则可以进行compact合并，最小的可合并数通过hbase.hstore.compactionThreshold配置，默认是3，最小值为2。

public boolean needsCompaction() {

    return (storefiles.size() - filesCompacting.size()) > minFilesToCompact;

}

 

//是否是major合并

private boolean isMajorCompaction(final List<StoreFile> filesToCompact) throws IOException {

boolean result = false;

//根据hbase.hregion.majorcompaction配置的major合并周期计算下次进行major合并的时间，如果设置为0则不进行major合并

long mcTime = getNextMajorCompactTime();

if (filesToCompact == null || filesToCompact.isEmpty() || mcTime == 0) {

      return result;

    }

// TODO: Use better method for determining stamp of last major (HBASE-2990)

//store中最久没有被修改过的storeFile文件的时间，作为上次major合并的时间进行判断下次应该进行major合并的时间，这种做法并不合理，可能会导致延后执行major合并，极端情况下会导致永远不进行major合并。

    long lowTimestamp = getLowestTimestamp(filesToCompact);

    long now = System.currentTimeMillis();

    //只有当达到了major合并时间才可能进行major合并

    if (lowTimestamp > 0l && lowTimestamp < (now - mcTime)) { 

      // Major compaction time has elapsed.

      if (filesToCompact.size() == 1) {

        StoreFile sf = filesToCompact.get(0);

        //store中最久的时间与当前时间的时间差

        long oldest = (sf.getReader().timeRangeTracker == null) ?

                Long.MIN_VALUE :

                now - sf.getReader().timeRangeTracker.minimumTimestamp;

        if (sf.isMajorCompaction() && (this.ttl == HConstants.FOREVER || oldest < this.ttl)) {

          //如果列簇没有设置过期时间(通过HColumnDescriptor.setTimeToLive()设置)，因此无需通过major合并删除过期数据。

          }

        } else if (this.ttl != HConstants.FOREVER && oldest > this.ttl) {

            result = true;

        }

      } else {

        result = true;

      }

    }

    return result;

  }

        

    2） 、表级

   通过HBaseAdmin或者CompactionTool可以触发表下的所有region和列簇进行compact合并（minor或者major）。HBaseAdmin还可以触发表下的指定列簇的compact操作。

       

    3）、region级

    通过HBaseAdmin或者CompactionTool可触发对指定region下的所有列簇进行compact操作（minor或者major）。HBaseAdmin还可以触发region下的指定列簇的compact操作。

   通过Merge工具可以把给定表下的任意两个region合并成一个region，在合并region前会触发region的major compact操作。

  在flush memstore过程中会触发当前region的compact，写数据或者split region等会触发flush memstore。

    

  4）、列簇级（Store级）

  有很多情况均会触发Store的compact，比如：执行CompactionTool工具的compact方式、flush memstore等。

  注：以上4条只是指触发compact操作，但是不一定真正发生compact，还需满足needsCompaction()或者isMajorCompaction()的条件。

compact总结：

  1）、从compact的程度可以分为：minor和major合并；  

  2）、从发生的范围可以分：整个集群、表、region、列簇4个级别；

  3）、从触发的方式上可以分：

a、hbase内部自动触发（HRegionServer的定时器、flush memstore、split等）

b、客户端等外部触发（hbase管理工具、HBaseAdmin（client端管理类）、CompactionTool等）  

  4）、从执行的实时性：异步执行，立即执行；

 

  Compact的执行逻辑如下：

//CompactSplitThread类，只由HRegionServer类持有，在以下几个地方被调用：

//1、HRegionServer的compact守护线程

//2、MemStoreFlusher的flushRegion

//3、CompactingRequest的run方法

public synchronized CompactionRequest requestCompaction(final HRegion r, final Store s,

      final String why, int priority, CompactionRequest request) throws IOException {

…

    CompactionRequest cr = s.requestCompaction(priority, request);

…

      cr.setServer(server);

…

//是否是large合并，只与参与合并的文件的总大小有关，超过一定值后就会通过large合并的线程池，

//注意与major合并的区别，large线程池执行的任务可能是一个minor合并也可能是major合并。

//默认的large和small线程数是1，可以通过hbase.regionserver.thread.compaction.large和hbase.regionserver.thread.compaction.small配置

ThreadPoolExecutor pool = s.throttleCompaction(cr.getSize())? largeCompactions : smallCompactions;

      pool.execute(cr);

…

    return cr;

  }

 

//Store类

public CompactionRequest requestCompaction(int priority, CompactionRequest request)

      throws IOException {

   …

    this.lock.readLock().lock();

    try {

      synchronized (filesCompacting) {

        // candidates = all storefiles not already in compaction queue

        List<StoreFile> candidates = Lists.newArrayList(storefiles);

        if (!filesCompacting.isEmpty()) {

          // exclude all files older than the newest file we're currently

          // compacting. this allows us to preserve contiguity (HBASE-2856)

          StoreFile last = filesCompacting.get(filesCompacting.size() - 1);

          int idx = candidates.indexOf(last);

          Preconditions.checkArgument(idx != -1);

          candidates.subList(0, idx + 1).clear();

        }

 

        boolean override = false;

        if (region.getCoprocessorHost() != null) {

          override = region.getCoprocessorHost().preCompactSelection(this, candidates, request);

        }

        CompactSelection filesToCompact;

        if (override) {

          // coprocessor is overriding normal file selection

          filesToCompact = new CompactSelection(conf, candidates);

        } else {

          filesToCompact = compactSelection(candidates, priority);

        }

 

        if (region.getCoprocessorHost() != null) {

          region.getCoprocessorHost().postCompactSelection(this,

            ImmutableList.copyOf(filesToCompact.getFilesToCompact()), request);

        }

 

        // no files to compact

        if (filesToCompact.getFilesToCompact().isEmpty()) {

          return null;

        }

 

        // basic sanity check: do not try to compact the same StoreFile twice.

        if (!Collections.disjoint(filesCompacting, filesToCompact.getFilesToCompact())) {

          // TODO: change this from an IAE to LOG.error after sufficient testing

          Preconditions.checkArgument(false, "%s overlaps with %s",

              filesToCompact, filesCompacting);

        }

        filesCompacting.addAll(filesToCompact.getFilesToCompact());

        Collections.sort(filesCompacting, StoreFile.Comparators.FLUSH_TIME);

 

        // major compaction iff all StoreFiles are included

        boolean isMajor = (filesToCompact.getFilesToCompact().size() == this.storefiles.size());

        if (isMajor) {

          // since we're enqueuing a major, update the compaction wait interval

          this.forceMajor = false;

        }

 

        // everything went better than expected. create a compaction request

        int pri = getCompactPriority(priority);

        //not a special compaction request, so we need to make one

        if(request == null){

          request = new CompactionRequest(region, this, filesToCompact, isMajor, pri);

        } else {

          // update the request with what the system thinks the request should be

          // its up to the request if it wants to listen

          request.setSelection(filesToCompact);

          request.setIsMajor(isMajor);

          request.setPriority(pri);

        }

      }

    } finally {

      this.lock.readLock().unlock();

    }

    if (request != null) {

      CompactionRequest.preRequest(request);

    }

    return request;

  }

 

//如果合并的总文件大小超过2 * this.minFilesToCompact * this.region.memstoreFlushSize则会通过大合并的线程池进行合并，总共有两个合并的线程池

ThreadPoolExecutor pool = s.throttleCompaction(cr.getSize())? largeCompactions : smallCompactions;

 

// minFilesToCompact默认值为3， memstoreFlushSize默认值128M

boolean throttleCompaction(long compactionSize) {

    long throttlePoint = conf.getLong(

        "hbase.regionserver.thread.compaction.throttle",

        2 * this.minFilesToCompact * this.region.memstoreFlushSize);

    return compactionSize > throttlePoint;

  }

 

 

 4、Split

  HBase的默认split策略类是：IncreasingToUpperBoundRegionSplitPolicy，可以通过hbase.regionserver.region.split.policy配置，或者通过HTableDescriptor在建表时指，HTableDescriptor指定的split策略优先级最高，以下是对该类中计算split临界大小的源代码讲解：

 

//IncreasingToUpperBoundRegionSplitPolicy类

//返回需要split的storeFile大小，如果超过该值，则可能触发split操作

//取region数量和memstore大小的计算值与desiredMaxFileSize比较的最小值，因此在进行写数据时，我们会发现虽然配置的最大region大小为10G，但是hbase并不会真正等region大小达到10G才split，而是有各种split的触发大小，当只有一个region时，达到memstore大小就会split，如此设计可以确保写数据时可以快速分裂出多个region，充分利用集群资源，并且在早期split会比中后期进行split消耗的服务器资源更少，因为早期数据量小。

  long getSizeToCheck(final int tableRegionsCount) {

    return tableRegionsCount == 0? getDesiredMaxFileSize():

      Math.min(getDesiredMaxFileSize(),

        this.flushSize * (tableRegionsCount * tableRegionsCount));

  }

// getDesiredMaxFileSize的逻辑如下：

//如果建表时指定了region大小，则采用建表时指定的值，否则采用hbase.hregion.max.filesize配置的值

HTableDescriptor desc = region.getTableDesc();

    if (desc != null) {

      this.desiredMaxFileSize = desc.getMaxFileSize();

    }

    if (this.desiredMaxFileSize <= 0) {

      this.desiredMaxFileSize = conf.getLong(HConstants.HREGION_MAX_FILESIZE,

        HConstants.DEFAULT_MAX_FILE_SIZE);

    }