hbase源码系列（十四）Compact和Split

先上一张图讲一下Compaction和Split的关系，这样会比较直观一些。

Compaction把多个MemStore flush出来的StoreFile合并成一个文件，而Split则是把过大的文件Split成两个。

之前在Delete的时候，我们知道它其实并没有真正删除数据的，那总不能一直不删吧，下面我们就介绍一下它删除数据的过程，它就是Compaction。

在讲源码之前，先说一下它的分类和作用。

Compaction主要起到如下几个作用：

1）合并文件

2）清除删除、过期、多余版本的数据

3）提高读写数据的效率

Minor & Major Compaction的区别

1）Minor操作只用来做部分文件的合并操作以及包括minVersion=0并且设置ttl的过期版本清理，不做任何删除数据、多版本数据的清理工作。

2）Major操作是对Region下的HStore下的所有StoreFile执行合并操作，最终的结果是整理合并出一个文件。

说了那么多，我们开始看入口吧，入口在HBaseAdmin，找到compact方法，都知道我们compact可以对表操作或者对region进行操作。

1、先把表或者region相关的region信息和server信息全部获取出来

2、循环遍历这些region信息，依次请求compact操作


AdminService.BlockingInterface admin = this.connection.getAdmin(sn);

CompactRegionRequest request = RequestConverter.buildCompactRegionRequest(hri.getRegionName(), major, family);

try {

      admin.compactRegion(null, request);

} catch (ServiceException se) {

      throw ProtobufUtil.getRemoteException(se);

}

到这里，客户端的工作就结束了，我们直接到HRegionServer找compactRegion这个方法吧。


　　　 //major compaction多走这一步骤

      if (major) {

        if (family != null) {

          store.triggerMajorCompaction();

        } else {

          region.triggerMajorCompaction();

        }

      }

　　　　//请求compaction走这里

      if(family != null) {

        compactSplitThread.requestCompaction(region, store, log, Store.PRIORITY_USER, null);

      } else {

        compactSplitThread.requestCompaction(region, log, Store.PRIORITY_USER, null);

      }

我们先看major
compaction吧，直接去看triggerMajorCompaction和requestCompaction方法。

Compaction

进入方法里面就发现了它把forceMajor置为true就完了，看来这个参数是major和minor的开关，接着看requestCompaction。


CompactionContext compaction = null;

if (selectNow) {

    compaction = selectCompaction(r, s, priority, request);

    if (compaction == null) return null; // message logged inside

}

// 要根据文件的size来判断用给个大的线程池还是小的线程池

long size = selectNow ? compaction.getRequest().getSize() : 0;

ThreadPoolExecutor pool = (!selectNow && s.throttleCompaction(size)) ? largeCompactions : smallCompactions;

pool.execute(new CompactionRunner(s, r, compaction, pool));

上面的步骤是执行selectCompaction创建一个CompactionContext，然后提交CompactionRunner。

我们接着看CompactionContext的创建过程吧，这里还需要分是用户创建的Compaction和系统创建的Compaction。

1、创建CompactionContext


2、判断是否是非高峰时间，下面是这两个参数的值

int startHour = conf.getInt("hbase.offpeak.start.hour", -1);

int endHour = conf.getInt("hbase.offpeak.end.hour", -1);

3、选择需要进行compaction的文件，添加到CompactionRequest和filesCompacting列表当中

compaction.select(this.filesCompacting, isUserCompaction, mayUseOffPeak, forceMajor && filesCompacting.isEmpty());

我们看看这个select的具体实现吧。

public boolean select(List<StoreFile> filesCompacting, boolean isUserCompaction,

        boolean mayUseOffPeak, boolean forceMajor) throws IOException {

      request = compactionPolicy.selectCompaction(storeFileManager.getStorefiles(),

          filesCompacting, isUserCompaction, mayUseOffPeak, forceMajor);

      return request != null;

}

这里的select方法，从名字上看是压缩策略的意思，它是由这个参数控制的hbase.hstore.defaultengine.compactionpolicy.class，默认是ExploringCompactionPolicy这个类。

接着看ExploringCompactionPolicy的selectCompaction方法，发现这个方法是继承来的，找它的父类RatioBasedCompactionPolicy。

public CompactionRequest selectCompaction(Collection<StoreFile> candidateFiles,

      final List<StoreFile> filesCompacting, final boolean isUserCompaction,

      final boolean mayUseOffPeak, final boolean forceMajor) throws IOException {

    ArrayList<StoreFile> candidateSelection = new ArrayList<StoreFile>(candidateFiles);

    int futureFiles = filesCompacting.isEmpty() ? 0 : 1;

    boolean mayBeStuck = (candidateFiles.size() - filesCompacting.size() + futureFiles)

        >= storeConfigInfo.getBlockingFileCount();

    //从candidateSelection排除掉filesCompacting中的文件

    candidateSelection = getCurrentEligibleFiles(candidateSelection, filesCompacting);long cfTtl = this.storeConfigInfo.getStoreFileTtl();

    if (!forceMajor) {

      // 如果不是强制major的话，包含了过期的文件，先删除过期的文件

      if (comConf.shouldDeleteExpired() && (cfTtl != Long.MAX_VALUE)) {

        ArrayList<StoreFile> expiredSelection = selectExpiredStoreFiles(

            candidateSelection, EnvironmentEdgeManager.currentTimeMillis() - cfTtl);

        if (expiredSelection != null) {

          return new CompactionRequest(expiredSelection);

        }

      }

      //居然还要跳过大文件，看来不是major的还是不行的，净挑小的弄

      candidateSelection = skipLargeFiles(candidateSelection);

    }

    // 是不是major的compaction还需要判断，做这个操作还是比较谨慎的

    boolean majorCompaction = (

      (forceMajor && isUserCompaction)

      || ((forceMajor || isMajorCompaction(candidateSelection))

          && (candidateSelection.size() < comConf.getMaxFilesToCompact()))

      || StoreUtils.hasReferences(candidateSelection)

      );


if (!majorCompaction) {

　　   //过滤掉bulk load进来的文件

      candidateSelection = filterBulk(candidateSelection);

      //过滤掉一些不满足大小的文件

      candidateSelection = applyCompactionPolicy(candidateSelection, mayUseOffPeak, mayBeStuck);

      //检查文件数是否满足最小的要求，文件不够，也不做compaction

      candidateSelection = checkMinFilesCriteria(candidateSelection);

    }

    //非major的超过最大可以compact的文件数量也要剔除掉，major的只是警告一下

    candidateSelection = removeExcessFiles(candidateSelection, isUserCompaction, majorCompaction);

    CompactionRequest result = new CompactionRequest(candidateSelection);

    result.setOffPeak(!candidateSelection.isEmpty() && !majorCompaction && mayUseOffPeak);

    return result;

  }

从上面可以看出来，major compaction的选择文件几乎没什么限制，只要排除掉正在compacting的文件就行了，反而是minor
compact有诸多的排除选项，因为默认的compaction是定时执行的，所以它这方面的考虑吧，排除太大的文件，选择那些过期的文件，排除掉bulkload的文件等等内容。

Minor Compaction的文件选择策略

我们再简单看看applyCompactionPolicy这个方法吧，它是minor的时候用的，它的过程就像下图一样。

这个是双层循环：

从0开始，循环N遍（N=文件数），就相当于窗口向右滑动，指针为start

----->从currentEnd=start + MinFiles（默认是3）-1，每次增加一个文件作为考虑，类似扩张的动作, 窗口扩大,
指针为

-------------->从candidateSelection文件里面取出(start, currentEnd + 1)开始

-------------->小于最小compact数量文件，默认是3，continue

-------------->大于最大compact数量文件，默认是10，continue

-------------->获取这部分文件的大小

-------------->如果这部分文件数量比上次选择方案的文件还小，替换为最小文件方案

-------------->大于MemStore flush的大小128M并且符合有一个文件不满这个公式（FileSize(i) <= (
文件总大小- FileSize(i) ) * Ratio），continue

　　　　　　　（注意上面的Ratio是干嘛的，这个和前面提到的非高峰时间的数值有关系，非高峰时段这个数值是5，高峰时间段这个值是1.2,
这说明高峰时段不允许compact过大的文件）

-------------->开始判断是不是最优的选择(下面讲的mayBeStuck是从selectCompaction传入的，可选择的文件超过7个的情况，上面黄色那部分代码）

　　　　　　　　　　1）如果mayBeStuck并且不是初次，如果 文件平均大小 > 上次选择的文件的平均大小*1.05,
替换上次的选择文件方案成为最优解

　　　　　　　　　　2）初次或者不是mayBeStuck的情况，文件更多的或者文件相同、总文件大小更小的会成为最新的选择文件方案

如果经过比较之后的最优文件选择方案不为空，就把它返回，否则就把最小文件方案返回。

下面是之前的Ratio的参数值，需要配合之前提到的参数配合使用的。

hbase.hstore.compaction.ratio              高峰时段，默认值是1.2

hbase.hstore.compaction.ratio.offpeak      非高峰时段，默认值是5

到这里先来个小结吧，从上面可以看得出来，这个Minor Compaction的文件选择策略就是选小的来，选最多的小文件来合并。

选择文件结束，回到compact的主流程


4、把CompactionRequest放入CompactionRunner，走线程池提交

之前的代码我再贴一下，省得大家有点凌乱。

ThreadPoolExecutor pool = (!selectNow && s.throttleCompaction(size)) ? largeCompactions : smallCompactions;

pool.execute(new CompactionRunner(s, r, compaction, pool));

我们去看CompactionRunner的run方法吧，它也在当前的类里面。


      if (this.compaction == null) {this.compaction = selectCompaction(this.region, this.store, queuedPriority, null); 
　　　　 // 出口，实在选不出东西来了，它会走这里跑掉
　　　　 if (this.compaction == null) return;
　　　　 // ....还有别的限制，和父亲运行的线程池也要一致，尼玛，什么逻辑  
　　　 }

        
　　　　boolean completed = region.compact(compaction, store);if (completed) {

       // blocked的regions再来一次，这次又要一次compaction意欲何为啊
       // 其实它的出口在上面的那段代码，它执行之后，没有这里这么恶心
       if (store.getCompactPriority() <= 0) {

           requestSystemCompaction(region, store, "Recursive enqueue");

       } else {

         // compaction之后的region可能很大，超过split的数量就要split了
         requestSplit(region);

       }

先是对region进行compact，如果完成了，判断一下优先级，优先级小于等于0，请求系统级别的compaction，否则请求split。

我们还是先看HRegion的compact方法，compact开始前，它要先上读锁，不让读了，然后调用HStore中的compact方法。


　　   // 执行compact，生成新文件

      List<Path> newFiles = compaction.compact();

      //把compact生成的文件移动到正确的位置

      sfs = moveCompatedFilesIntoPlace(cr, newFiles);

      //记录WALEdit日志

      writeCompactionWalRecord(filesToCompact, sfs);

      //更新HStore相关的数据结构

      replaceStoreFiles(filesToCompact, sfs);/
      /归档旧的文件，关闭reader，重新计算file的大小

      completeCompaction(filesToCompact);

comact生成新文件的方法很简单，给源文件创建一个StoreScanner，之前说过StoreScanner能从多个Scanner当中每次都取出最小的kv，然后用StoreFile.Append的方法不停地追加写入即可，这些过程在前面的章节都介绍过了，这里不再重复。

简单的说，就是把这些文件合并到一个文件去了，尼玛，怪不得io那么大。

剩下的就是清理工作了，这里面有意思的就是它会记录一笔日志到writeCompactionWalRecord当中，在之间日志恢复那一章的时候，贴出来的代码里面有，只是没有详细的讲。因为走到这里它已经完成了compaction的过程，只是没有把旧的文件移入归档文件当中，它挂掉重启的时候进行恢复干的事情，就是替换文件。

5、store.getCompactPriority() 下一步是天堂抑或是地狱？

compact完了，要判断一下这个，真是天才啊

public int getStoreCompactionPriority() {

    int blockingFileCount = conf.getInt(

        HStore.BLOCKING_STOREFILES_KEY, HStore.DEFAULT_BLOCKING_STOREFILE_COUNT);

    int priority = blockingFileCount - storefiles.size();

    return (priority == HStore.PRIORITY_USER) ? priority + 1 : priority;

}

比较方法是这个，blockingFileCount的默认值是7，如果compact之后storefiles的文件数量大于7的话，就很有可能再触发一下，那么major
compaction触发的可能性低，minor触发的可能性非常大。

不过没关系，实在选不出文件来，它会退出的。咱们可以将它这个参数hbase.hstore.blockingStoreFiles设置得大一些，弄出来一个比较大的数字，因为它还会影响到MemStore的flush，超过了那么多文件，它就会停止等待，所以它才叫停止等待文件数量。

这里还是建议没事的时候，把compaction给禁用了，如果等它自己执行，IO太大了会对线上应用影响的。

Split

好，我们接着看requestSplit。


if (shouldSplitRegion() && r.getCompactPriority() >= Store.PRIORITY_USER) {

      byte[] midKey = r.checkSplit();

      if (midKey != null) {

        requestSplit(r, midKey);

        return true;

      }

}

先检查一下是否可以进行split，如果可以，把中间的key返回来。

那条件是啥？在这里，if的条件是成立的，条件判断在IncreasingToUpperBoundRegionSplitPolicy的shouldSplit方法当中。

遍历region里面所有的store

1、Store当中不能有Reference文件。

2、store.size > Math.min(getDesiredMaxFileSize(), this.flushSize *
(tableRegionsCount * (long)tableRegionsCount)) 就返回ture，可以split。

getDesiredMaxFileSize()默认是10G，由这个参数来确定hbase.hregion.max.filesize,
当没超过10G的时候它就会根据128MB * （该表在这个RS上的region数量）平方。

midKey怎么找呢？找出最大的HStore，然后通过它来找这个分裂点，最大的文件的中间点。

return StoreUtils.getLargestFile(this.storefiles).getFileSplitPoint(this.kvComparator);

但是如果是另外一种情况，我们通过客户端来分裂Region，我们强制指定的分裂点，这种情况是按照我们设置的分裂点来进行分裂。

分裂点有了，我们接着看，我们发现它又提交了一个SplitRequest线程，看run方法。

1、先获得一个tableLock，给这个表上锁

2、执行SplitTransaction的prepare方法，然后execute

3、结束了释放tableLock


      // 先做准备工作，然后再execute执行主流程，过程当中出错了，就rollback
      if (!st.prepare()) return;

      try {

        st.execute(this.server, this.server);

      } catch (Exception e) {

        try {

        if (st.rollback(this.server, this.server)) {

        } catch (RuntimeException ee) {this.server.abort(msg);

        }

        return;

      }

prepare方法当中，主要做了这么件事，new了两个新的region出来

this.hri_a = new HRegionInfo(hri.getTable(), startKey, this.splitrow, false, rid);

this.hri_b = new HRegionInfo(hri.getTable(), this.splitrow, endKey, false, rid);

我们接着看execute方法，这个是重头戏。

PairOfSameType<HRegion> regions = createDaughters(server, services);

openDaughters(server, services, regions.getFirst(), regions.getSecond());

transitionZKNode(server, services, regions.getFirst(), regions.getSecond());

总共分三步：

1、创建子region

2、上线子region

3、更改zk当中的状态

我们先看createDaughters


    //在region-in-transition节点下给父region创建一个splitting的节点
    createNodeSplitting(server.getZooKeeper(), parent.getRegionInfo(), server.getServerName(), hri_a, hri_b);

    this.journal.add(JournalEntry.SET_SPLITTING_IN_ZK);//在parent的region目录下创建.splits目录

    this.parent.getRegionFileSystem().createSplitsDir();

    this.journal.add(JournalEntry.CREATE_SPLIT_DIR);


Map<byte[], List<StoreFile>> hstoreFilesToSplit = null;//关闭parent，然后返回相应的列族和storefile的map

    hstoreFilesToSplit = this.parent.close(false);//从在线列表里下线parent

    services.removeFromOnlineRegions(this.parent, null);

    this.journal.add(JournalEntry.OFFLINED_PARENT);

// 把parent的storefile均分给两个daughter，所谓均分，只是创建引用文件而已

    splitStoreFiles(hstoreFilesToSplit);


// 把临时的Region A目录重名为正式的region A 的目录    
    this.journal.add(JournalEntry.STARTED_REGION_A_CREATION);

    HRegion a = this.parent.createDaughterRegionFromSplits(this.hri_a);


// 把临时的Region B目录重名为正式的region B的目录
    this.journal.add(JournalEntry.STARTED_REGION_B_CREATION);

    HRegion b = this.parent.createDaughterRegionFromSplits(this.hri_b);

    this.journal.add(JournalEntry.PONR);


// 修改meta表中的信息，设置parent的状态为下线、并且split过，在增加两列左右孩子，左右孩子的信息也通过put插入到meta中

    MetaEditor.splitRegion(server.getCatalogTracker(), parent.getRegionInfo(),

          a.getRegionInfo(), b.getRegionInfo(), server.getServerName());

    return new PairOfSameType<HRegion>(a, b);

在splitStoreFiles这块的，它给每个文件都开一个线程去进行split。

fs.splitStoreFile(this.hri_a, familyName, sf, this.splitrow, false);

fs.splitStoreFile(this.hri_b, familyName, sf, this.splitrow, true);

这里其实是给每个文件都创建了Reference文件，无论它的文件当中包不包括splitRow。


    //parentRegion/.splits/region/familyName目录

    Path splitDir = new Path(getSplitsDir(hri), familyName);// 其实它并没有真正的split，而是通过创建Reference

    Reference r = top ? Reference.createTopReference(splitRow): Reference.createBottomReference(splitRow);

    String parentRegionName = regionInfo.getEncodedName();// 原来通过这么关联啊，storefile名字 + 父parent的name

    Path p = new Path(splitDir, f.getPath().getName() + "." + parentRegionName);

    return r.write(fs, p);

把引用文件生成在每个子region对应的目录，以便下一步直接重命令目录即可。

重命名目录之后，就是修改Meta表了，splitRegion的方法是通过Put来进行操作的，它修改parent的regioninfo这一列更新为最新的信息，另外又增加了splitA和splitB两列，hri_a和hri_b则通过另外两个Put插入到Meta表当中。

这个过程当中如果出现任何问题，就需要根据journal记录的过程信息进行回滚操作。

怎么open这两个子region就不讲了，之前讲《HMaster启动过程》的时候讲过了。

到这里split的过程就基本结束了。

hbase源码系列（十四）Compact和Split

时间： 2024-10-05 22:11:57

hbase源码系列（十四）Compact和Split的相关文章

hbase源码系列（十二）Get、Scan在服务端是如何处理？

继上一篇讲了Put和Delete之后,这一篇我们讲Get和Scan, 因为我发现这两个操作几乎是一样的过程,就像之前的Put和Delete一样,上一篇我本来只打算写Put的,结果发现Delete也可以走这个过程,所以就一起写了. Get 我们打开HRegionServer找到get方法.Get的方法处理分两种,设置了ClosestRowBefore和没有设置的,一般来讲,我们都是知道了明确的rowkey,不太会设置这个参数,它默认是false的. if (get.hasClosestRowBef

hbase源码系列（十）HLog与日志恢复

HLog概述 hbase在写入数据之前会先写入MemStore,成功了再写入HLog,当MemStore的数据丢失的时候,还可以用HLog的数据来进行恢复,下面先看看HLog的图. 旧版的HLog是实际上是一个SequceneFile,0.96的已经使用Protobuf来进行序列化了.从Writer和Reader上来看HLog的都是Entry的,换句话说就是,它的每一条记录就是一个Entry. class Entry implements Writable { private WALEdit e

hbase源码系列（二）HTable 如何访问客户端

hbase的源码终于搞一个段落了,在接下来的一个月,着重于把看过的源码提炼一下,对一些有意思的主题进行分享一下.继上一篇讲了负载均衡之后,这一篇我们从client开始讲吧,从client到master再到region server,按照这个顺序来开展,网友也可以对自己感兴趣的部分给我留言或者直接联系我的QQ. 现在我们讲一下HTable吧,为什么讲HTable,因为这是我们最常见的一个类,这是我们对hbase中数据的操作的入口. 1.Put操作 下面是一个很简单往hbase插入一条记录的例子.

hbase源码系列（十一）Put、Delete在服务端是如何处理？

在讲完之后HFile和HLog之后,今天我想分享是Put在Region Server经历些了什么?相信前面看了<HTable探秘>的朋友都会有印象,没看过的建议回去先看看,Put是通过MultiServerCallable来提交的多个Put,好,我们就先去这个类吧,在call方法里面,我们找到了这句. responseProto = getStub().multi(controller, requestProto); 它调用了Region Server的multi方法.好,我们立即杀到HReg

hbase源码系列（七）Snapshot的过程

在看这一章之前,建议大家先去看一下snapshot的使用.可能有人会有疑问为什么要做Snapshot,hdfs不是自带了3个备份吗,这是个很大的误区,要知道hdfs的3个备份是用于防止网络传输中的失败或者别的异常情况导致数据块丢失或者不正确,它不能避免人为的删除数据导致的后果.它就想是给数据库做备份,尤其是做删除动作之前,不管是hbase还是hdfs,请经常做Snapshot,否则哪天手贱了... 直接进入主题吧,上代码. public void takeSnapshot(SnapshotDes

hbase源码系列（三）Client如何找到正确的Region Server

客户端在进行put.delete.get等操作的时候,它都需要数据到底存在哪个Region Server上面,这个定位的操作是通过HConnection.locateRegion方法来完成的. loc = hConnection.locateRegion(this.tableName, row.getRow()); 这里我们首先要讲hbase的两张元数据表-ROOT-和.META.表,它们一个保存着region的分部信息,一个保存着region的详细信息.在<hbase实战>这本书里面详细写了

Spark源码系列（四）图解作业生命周期

这一章我们探索了Spark作业的运行过程,但是没把整个过程描绘出来,好,跟着我走吧,let you know! 我们先回顾一下这个图,Driver Program是我们写的那个程序,它的核心是SparkContext,回想一下,从api的使用角度,RDD都必须通过它来获得. 下面讲一讲它所不为认知的一面,它和其它组件是如何交互的. Driver向Master注册Application过程 SparkContext实例化之后,在内部实例化两个很重要的类,DAGScheduler和TaskSched

hbase源码系列（五）单词查找树

在上一章中提到了编码压缩,讲了一个简单的DataBlockEncoding.PREFIX算法,它用的是前序编码压缩的算法,它搜索到时候,是全扫描的方式搜索的,如此一来,搜索效率实在是不敢恭维,所以在hbase当中单独拿了一个工程出来实现了Trie的数据结果,既达到了压缩编码的效果,亦达到了方便查询的效果,一举两得,设置的方法是在上一章的末尾提了. 下面讲一下这个Trie树的原理吧. hbase源码系列(五)单词查找树,布布扣,bubuko.com

hbase源码系列（十三）缓存机制MemStore与Block Cache

这一章讲hbase的缓存机制,这里面涉及的内容也是比较多,呵呵,我理解中的缓存是保存在内存中的特定的便于检索的数据结构就是缓存. 之前在讲put的时候,put是被添加到Store里面,这个Store是个接口,实现是在HStore里面,MemStore其实是它底下的小子. 那它和Region Server.Region是什么关系? Region Server下面有若干个Region,每个Region下面有若干的列族,每个列族对应着一个HStore. HStore里面有三个很重要的类,在这章的内容都