图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Hbase › 代码分析› hbase源码Compact和Split
分享

hbase源码Compact和Split

xioaxu790 2014-10-17 13:06:50 发表于 代码分析 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 0 13061
About云VIP帮助找工作和提升
问题导读
1、你如何理解Compaction和Split的关系？
2、Minor 和Major Compaction的区别有哪些？
3、splitStoreFiles这块有什么作用？






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


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执行合并操作，最终的结果是整理合并出一个文件。

先说一下怎么使用吧，下面分别是它们是shell命令，可以在hbase的shell里面执行。
  1. //major compaction

  2. major compact '表名或region名'

  3. 

  4. //minor compaction

  5. compact '表名或region名'
复制代码


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

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

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

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

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

  3. try {

  4.       admin.compactRegion(null, request);

  5. } catch (ServiceException se) {

  6.       throw ProtobufUtil.getRemoteException(se);

  7. }
复制代码

到这里，客户端的工作就结束了，我们直接到HRegionServer找compactRegion这个方法吧。
  1. 　　　 //major compaction多走这一步骤

  2.       if (major) {

  3.         if (family != null) {

  4.           store.triggerMajorCompaction();

  5.         } else {

  6.           region.triggerMajorCompaction();

  7.         }

  8.       }

  9. 　　　　//请求compaction走这里

  10.       if(family != null) {

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

  12.       } else {

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

  14.       }
复制代码


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

Compaction

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

  1. CompactionContext compaction = null;

  2. if (selectNow) {

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

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

  5. }

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

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

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

  9. pool.execute(new CompactionRunner(s, r, compaction, pool));
复制代码

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

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

1、创建CompactionContext
2、判断是否是非高峰时间，下面是这两个参数的值
  1. int startHour = conf.getInt("hbase.offpeak.start.hour", -1);

  2. int endHour = conf.getInt("hbase.offpeak.end.hour", -1);
复制代码


3、选择需要进行compaction的文件，添加到CompactionRequest和filesCompacting列表当中
compaction.select(this.filesCompacting, isUserCompaction, mayUseOffPeak, forceMajor && filesCompacting.isEmpty());
我们看看这个select的具体实现吧。
  1. public boolean select(List<StoreFile> filesCompacting, boolean isUserCompaction,

  2.         boolean mayUseOffPeak, boolean forceMajor) throws IOException {

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

  4.           filesCompacting, isUserCompaction, mayUseOffPeak, forceMajor);

  5.       return request != null;

  6. }
复制代码


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

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

  1. public CompactionRequest selectCompaction(Collection<StoreFile> candidateFiles,

  2.       final List<StoreFile> filesCompacting, final boolean isUserCompaction,

  3.       final boolean mayUseOffPeak, final boolean forceMajor) throws IOException {

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

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

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

  7.         >= storeConfigInfo.getBlockingFileCount();

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

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

  10.     if (!forceMajor) {

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

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

  13.         ArrayList<StoreFile> expiredSelection = selectExpiredStoreFiles(

  14.             candidateSelection, EnvironmentEdgeManager.currentTimeMillis() - cfTtl);

  15.         if (expiredSelection != null) {

  16.           return new CompactionRequest(expiredSelection);

  17.         }

  18.       }

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

  20.       candidateSelection = skipLargeFiles(candidateSelection);

  21.     }

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

  23.     boolean majorCompaction = (

  24.       (forceMajor && isUserCompaction)

  25.       || ((forceMajor || isMajorCompaction(candidateSelection))

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

  27.       || StoreUtils.hasReferences(candidateSelection)

  28.       );

  29. 

  30.     if (!majorCompaction) {

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

  32.       candidateSelection = filterBulk(candidateSelection);

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

  34.       candidateSelection = applyCompactionPolicy(candidateSelection, mayUseOffPeak, mayBeStuck);

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

  36.       candidateSelection = checkMinFilesCriteria(candidateSelection);

  37.     }

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

  39.     candidateSelection = removeExcessFiles(candidateSelection, isUserCompaction, majorCompaction);

  40.     CompactionRequest result = new CompactionRequest(candidateSelection);

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

  42.     return result;

  43.   }
复制代码

从上面可以看出来，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，走线程池提交
之前的代码我再贴一下，省得大家有点凌乱。
  1. ThreadPoolExecutor pool = (!selectNow && s.throttleCompaction(size)) ? largeCompactions : smallCompactions;

  2. pool.execute(new CompactionRunner(s, r, compaction, pool));
复制代码


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

  1.       if (this.compaction == null) {this.compaction = selectCompaction(this.region, this.store, queuedPriority, null); 

  2. 　　　　 // 出口，实在选不出东西来了，它会走这里跑掉

  3. 　　　　 if (this.compaction == null) return;

  4. 　　　　 // ....还有别的限制，和父亲运行的线程池也要一致，尼玛，什么逻辑  

  5. 　　　 }

  6.         

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

  8.        // blocked的regions再来一次，这次又要一次compaction意欲何为啊

  9.        // 其实它的出口在上面的那段代码，它执行之后，没有这里这么恶心

  10.        if (store.getCompactPriority() <= 0) {

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

  12.        } else {

  13.          // compaction之后的region可能很大，超过split的数量就要split了

  14.          requestSplit(region);

  15.        }
复制代码

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

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

  1. 　　   // 执行compact，生成新文件

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

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

  4.       sfs = moveCompatedFilesIntoPlace(cr, newFiles);

  5.       //记录WALEdit日志

  6.       writeCompactionWalRecord(filesToCompact, sfs);

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

  8.       replaceStoreFiles(filesToCompact, sfs);/

  9.       /归档旧的文件，关闭reader，重新计算file的大小

  10.       completeCompaction(filesToCompact);
复制代码

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

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

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

5、store.getCompactPriority() 下一步是天堂抑或是地狱？
compact完了，要判断一下这个，真是天才啊
  1. public int getStoreCompactionPriority() {

  2.     int blockingFileCount = conf.getInt(

  3.         HStore.BLOCKING_STOREFILES_KEY, HStore.DEFAULT_BLOCKING_STOREFILE_COUNT);

  4.     int priority = blockingFileCount - storefiles.size();

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

  6. }
复制代码


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

不过没关系，实在选不出文件来，它会退出的。咱们可以将它这个参数hbase.hstore.blockingStoreFiles设置得大一些，弄出来一个比较大的数字。

Split
好，我们接着看requestSplit。
  1. if (shouldSplitRegion() && r.getCompactPriority() >= Store.PRIORITY_USER) {

  2.       byte[] midKey = r.checkSplit();

  3.       if (midKey != null) {

  4.         requestSplit(r, midKey);

  5.         return true;

  6.       }

  7. }
复制代码


先检查一下是否可以进行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
  1.       // 先做准备工作，然后再execute执行主流程，过程当中出错了，就rollback

  2.       if (!st.prepare()) return;

  3.       try {

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

  5.       } catch (Exception e) {

  6.         try {

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

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

  9.         }

  10.         return;

  11.       }
复制代码


prepare方法当中，主要做了这么件事，new了两个新的region出来
  1. this.hri_a = new HRegionInfo(hri.getTable(), startKey, this.splitrow, false, rid);

  2. this.hri_b = new HRegionInfo(hri.getTable(), this.splitrow, endKey, false, rid);
复制代码


我们接着看execute方法，这个是重头戏。
  1. PairOfSameType<HRegion> regions = createDaughters(server, services);

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

  3. transitionZKNode(server, services, regions.getFirst(), regions.getSecond());
复制代码


总共分三步：
1、创建子region

2、上线子region

3、更改zk当中的状态

我们先看createDaughters
  1.     //在region-in-transition节点下给父region创建一个splitting的节点

  2.     createNodeSplitting(server.getZooKeeper(), parent.getRegionInfo(), server.getServerName(), hri_a, hri_b);

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

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

  5.     this.journal.add(JournalEntry.CREATE_SPLIT_DIR);

  6. 

  7.     Map<byte[], List<StoreFile>> hstoreFilesToSplit = null;

  8.     //关闭parent，然后返回相应的列族和storefile的map

  9.     hstoreFilesToSplit = this.parent.close(false);

  10.     //从在线列表里下线parent

  11.     services.removeFromOnlineRegions(this.parent, null);

  12.     this.journal.add(JournalEntry.OFFLINED_PARENT);

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

  14.     splitStoreFiles(hstoreFilesToSplit);

  15. 

  16.     // 把临时的Region A目录重名为正式的region A 的目录    

  17.     this.journal.add(JournalEntry.STARTED_REGION_A_CREATION);

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

  19. 

  20.     // 把临时的Region B目录重名为正式的region B的目录

  21.     this.journal.add(JournalEntry.STARTED_REGION_B_CREATION);

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

  23.     this.journal.add(JournalEntry.PONR);

  24. 

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

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

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

  28.     return new PairOfSameType<HRegion>(a, b);
复制代码


在splitStoreFiles这块的，它给每个文件都开一个线程去进行split。
  1. fs.splitStoreFile(this.hri_a, familyName, sf, this.splitrow, false);

  2. fs.splitStoreFile(this.hri_b, familyName, sf, this.splitrow, true);
复制代码


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

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

  2.     Path splitDir = new Path(getSplitsDir(hri), familyName);

  3.     // 其实它并没有真正的split，而是通过创建Reference

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

  5.     String parentRegionName = regionInfo.getEncodedName();

  6.     // 原来通过这么关联啊，storefile名字 + 父parent的name

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

  8.     return r.write(fs, p);
复制代码

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

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

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

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

到这里split的过程就基本结束了，鉴于Compaction和Split的对io方面的巨大影响，所以在任何资料里面都是推荐屏蔽自动执行，写脚本在晚上自动进行这些操作。

关注公众号，获取大数据、人工智能20套、区块链资源5阶段等资源，随时更新，获取最新技术资源
回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
xioaxu790

金牌会员

关注

  • 851

    主题

  • 1000

    帖子

  • 108

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000