设为首页 加入收藏

TOP

HBase-压缩和分割原理
2019-03-05 01:43:05 【大 中 小】 浏览:56次
Tags：HBase- 压缩 分割 原理

HRegionServer调用合并请求

主要逻辑如下:

  1. //遍历每个Store然后计算需要合并的文件，生成
  2. //CompactionRequest对象并提交到线程池中执行
  3. //根据throttleCompaction()函数规则来判断是提交到
  4. //largeCompactions线程池还是smallCompactions线程池
  5. CompactSplitThread#requestCompaction(){
  6. for(Stores:r.getStores().values()){
  7. CompactionRequestcr=Store.requestCompaction(priority,request);
  8. ThreadPoolExecutorpool=s.throttleCompaction(cr.getSize())
  9. largeCompactions:smallCompactions;
  10. pool.execute(cr);
  11. ret.add(cr);
  12. }
  13. }
  15. //如果CompactionRequest的总大小>
  16. //minFilesToCompact*2*memstoreFlushSize
  17. //则这次任务为major合并，否则在为minor合并
  18. Store#throttleCompaction(){
  19. longthrottlePoint=conf.getLong(
  20. "hbase.regionserver.thread.compaction.throttle",
  21. 2*this.minFilesToCompact*this.region.memstoreFlushSize);
  22. returncompactionSize>throttlePoint;
  23. }
  26. Store#compactSelection(){
  27. //选择出已经过期的StoreFile
  28. if(storefile.maxTimeStamp+store.ttl<now_timestamp){
  29. //返回已经过期的storefile文件集合
  30. }
  32. //从0开始遍历到最后，如果发现有文件>maxCompactSize则pos++
  33. //然后过滤掉这些大于maxCompactSize的文件
  34. while(pos<compactSelection.getFilesToCompact().size()&&
  35. compactSelection.getFilesToCompact().get(pos).getReader().length()
  36. >while(pos<compactSelection.getFilesToCompact().size()&&
  37. compactSelection.getFilesToCompact().get(pos).getReader().length()
  38. >maxCompactSize&&
  39. !compactSelection.getFilesToCompact().get(pos).isReference())++pos;
  40. if(pos!=0)compactSelection.clearSubList(0,pos);&&
  41. !compactSelection.getFilesToCompact().get(pos).isReference()){
  42. ++pos;
  43. }
  44. if(pos!=0){
  45. compactSelection.clearSubList(0,pos);
  46. }
  47. if(compactSelection.getFilesToCompact().size()<minFilesToCompact){
  48. return;
  49. }
  51. //计算出sumSize数组，数组大小就是Store中的文件数量
  52. //sumSize数组中每个元素的大小是根据StroeFile的大小再加上sumSize[i+1](或者0)
  53. //然后减去fileSizes[tooFar](或者0)
  54. //sumSize的内容跟元素的fileSizes数组应该差别不大
  55. intcountOfFiles=compactSelection.getFilesToCompact().size();
  56. long[]fileSizes=newlong[countOfFiles];
  57. long[]sumSize=newlong[countOfFiles];
  58. for(inti=countOfFiles-1;i>=0;--i){
  59. StoreFilefile=compactSelection.getFilesToCompact().get(i);
  60. fileSizes[i]=file.getReader().length();
  61. //calculatethesumoffileSizes[i,i+maxFilesToCompact-1)foralgo
  62. inttooFar=i+this.maxFilesToCompact-1;
  63. sumSize[i]=fileSizes[i]+((i+1<countOfFiles)sumSize[i+1]:0)
  64. -((tooFar<countOfFiles)fileSizes[tooFar]:0);
  65. }
  67. //如果fileSize[start]>Math.max(minCompactSize,sumSize[start+1]*r)
  68. //则下标++，这里的操作是过滤掉过大的文件，以免影响合并时间
  69. while(countOfFiles-start>=this.minFilesToCompact&&fileSizes[start]>
  70. Math.max(minCompactSize,(long)(sumSize[start+1]*r))){
  71. ++start;
  72. }
  73. intend=Math.min(countOfFiles,start+this.maxFilesToCompact);
  74. longtotalSize=fileSizes[start]+((start+1<countOfFiles)sumSize[start+1]:0);
  75. compactSelection=compactSelection.getSubList(start,end);
  77. //如果是majorcompact，并且需要执行的文件数量过多，则去掉一些
  78. if(majorcompaction&&compactSelection.getFilesToCompact().size()>this.maxFilesToCompact){
  79. intpastMax=compactSelection.getFilesToCompact().size()-this.maxFilesToCompact;
  80. compactSelection.getFilesToCompact().subList(0,pastMax).clear();
  81. }
  82. }

CompactionRequest线程(用于执行major和minor合并)

压缩相关的类图如下:



major和minor合并的差别其实很小，如果最后待合并的总大小 > 2*minFilesToCompact*memstoreFlushSize

则认为这次是一个major合并，方到major线程池中执行，否则认为是一次minor合并

另外在创建StoreScanner构造函数时，会根据ScanType来判断是major还是minor合并，之后在

ScanQueryMathcer中根据ScanType的不同(有用户类型，minor和major三种类型)来决定返回的不同值的

主要逻辑如下:

  1. //在单独的线程中执行合并
  2. CompactionRequest#run(){
  3. booleancompleted=HRegion.compact(this);
  4. if(completed){
  5. if(s.getCompactPriority()<=0){
  6. server.getCompactSplitThread().requestCompaction(r,s,"Recursiveenqueue",null);
  7. }else{
  8. //seeifthecompactionhascausedustoexceedmaxregionsize
  9. server.getCompactSplitThread().requestSplit(r);
  10. }
  11. }
  12. }
  14. //这里会调用Store，来执行compact
  15. HRegion#compact(){
  16. Preconditions.checkArgument(cr.getHRegion().equals(this));
  17. lock.readLock().lock();
  18. CompactionRequest.getStore().compact(cr);
  19. lock.readLock().unlock();
  20. }
  22. //完成合并，调用Compactor#compact()完成最核心的compact逻辑
  23. //将合并后的文件移动到最终目录下并删除掉旧的文件
  24. Store#compact(){
  25. List<StoreFile>filesToCompact=request.getFiles();
  26. StoreFile.Writerwriter=this.compactor.compact(cr,maxId);
  27. if(this.conf.getBoolean("hbase.hstore.compaction.complete",true)){
  28. sf=completeCompaction(filesToCompact,writer);
  29. }else{
  30. //Createstorefilearoundwhatwewrotewithareaderonit.
  31. sf=newStoreFile(this.fs,writer.getPath(),this.conf,this.cacheConf,
  32. this.family.getBloomFilterType(),this.dataBlockEncoder);
  33. sf.createReader();
  34. }
  35. }
  37. //将/hbase/mytable/963cf86f3fd07c3d3161c1f4f15bef5a/.tmp/9c8614a6bd0d4833b419a13abfde5ac1
  38. //移动到
  39. ///hbase/mytable/963cf86f3fd07c3d3161c1f4f15bef5a/value/9c8614a6bd0d4833b419a13abfde5ac1
  40. //再对新的目标文件创建一个StroeFile对象包装
  41. //将旧的文件(这些底层的HFile都已经合并成一个文件了)删除
  42. //最后计算新的StoreFile文件大小等信息并返回
  43. Store#completeCompaction(){
  44. PathorigPath=compactedFile.getPath();
  45. PathdestPath=newPath(homedir,origPath.getName());
  46. HBaseFileSystem.renameDirForFileSystem(fs,origPath,destPath);
  47. StoreFileresult=newStoreFile(this.fs,destPath,this.conf,this.cacheConf,
  48. this.family.getBloomFilterType(),this.dataBlockEncoder);
  49. passSchemaMetricsTo(result);
  50. result.createReader();
  51. }
  54. //compact的最核心逻辑!!
  55. //对多个StoreFile进行合并，这里使用到了StoreScanner
  56. //迭代读取所有的StroeFile然后使用堆排序输出，并写入到
  57. //StoreFile$Writer#append()中
  58. Compactor#compact(){
  59. for(StoreFilefile:filesToCompact){
  60. StoreFile.Readerr=file.getReader();
  61. longkeyCount=(r.getBloomFilterType()==store.getFamily()
  62. .getBloomFilterType())
  63. r.getFilterEntries():r.getEntries();
  64. maxKeyCount+=keyCount;
  65. }
  67. intcompactionKVMax=getConf().getInt("hbase.hstore.compaction.kv.max",10);
  68. Compression.Algorithmcompression=store.getFamily().getCompression();
  69. List<StoreFileScanner>scanners=StoreFileScanner
  70. .getScannersForStoreFiles(filesToCompact,false,false,true);
  71. Scanscan=newScan();
  72. scan.setMaxVersions(store.getFamily().getMaxVersions());
  74. //这里会根据当前合并的类型选择ScanType的类型，之后ScanQueryMatcher根据ScanType的
  75. //的类型返回不同的值
  76. InternalScannerscanner=newStoreScanner(store,store.getScanInfo(),scan,scanners,majorCompactionScanType.<strong>MAJOR_COMPACT</strong>:ScanType.<strong>MINOR_COMPACT</strong>,
  77. smallestReadPoint,earliestPutTs);
  79. do{
  80. hasMore=scanner.next(kvs,compactionKVMax);
  81. if(writer==null&&!kvs.isEmpty()){
  82. //在tmp目录下创建一个临时文件，路径类似
  83. ///hbase/mytable/963cf86f3fd07c3d3161c1f4f15bef5a/.tmp/9c8614a6bd0d4833b419a13abfde5ac1
  84. writer=store.createWriterInTmp(maxKeyCount,compactionCompression,true,
  85. maxMVCCReadpoint>=smallestReadPoint);
  86. }
  87. for(KeyValuekv:kvs){
  88. writer.append(kv);
  89. }
  90. }while(hasMore);
  92. scanner.close();
  93. StoreFile$Writer.appendMetadata(maxId,majorCompaction);
  94. StoreFile$Writer.close();
  95. }

压缩算法和的核心逻辑演示类图

根据由新到老排序文件，选择出合适的文件

这里的滑动窗口是从0下标开始过滤掉size过大的文件，这样可以提高合并效率


使用到的一些重要类

其中内部scan的时候使用到的相关类图如下


相关重要的类:

  1. Hbase在实现该算法的过程中重要的是下面这五个类。
  2. 1.org.apache.hadoop.hbase.regionserver.Store
  3. 2.org.apache.hadoop.hbase.regionserver.StoreScanner
  4. 3.org.apache.hadoop.hbase.regionserver.StoreFileScanner
  5. 4.org.apache.hadoop.hbase.regionserver.KeyValueHeap
  6. 5.org.apache.hadoop.hbase.regionserver.ScanQueryMatcher
  8. 这五个类的关系是
  9. 1.Store类调用StoreScanner的next方法，并循环输出kv到合并文件；
  10. 2.StoreScanner的作用是负责创建并持有多个输入文件的StoreFileScanner，
  11. 内部遍历这些StoreFileScanner并通过KeyValueHeap来排序这些输入文件的首条记录；
  12. 3.StoreFileScanner的作用是遍历单个输入文件，管理并提供单个输入文件的首条记录；
  13. 4.KeyValueHeap的作用就是通过堆来排序每个输入文件的首条记录。
  14. 5.ScanQueryMatcher的作用是当输入文件的首条记录来的时候，根据一定的策略判断这条记录到底是该输出还是该跳过。

StoreScanner及相关类的主要逻辑如下:

  1. //内部应用StoreFileScanner列表，创建ScanQueryMatcher用来判断是过滤还是输出
  2. //创建KeyValueHeap用于堆排序，根据堆的结构每次从堆顶拿出一个
  3. //注意这个构造函数中有一个参数ScanType，是扫描的类型，包括MAJOR_COMPACT，MINOR_COMPACT，
  4. //USER_COMPACT来返回不同的值，以达到major或minor的效果
  5. StoreScanner#构造函数(){
  6. ScanQueryMatchermatcher=newScanQueryMatcher(scan,scanInfo,null,scanType,
  7. smallestReadPoint,earliestPutTs,oldestUnexpiredTS);
  8. List<extendsKeyValueScanner>scanners=selectScannersFrom(scanners);
  9. for(KeyValueScannerscanner:scanners){
  10. scanner.seek(matcher.getStartKey());
  11. }
  12. KeyValueHeapheap=newKeyValueHeap(scanners,store.comparator);
  13. }
  15. //选择性的创建布隆过滤器，调用HFileWriterv2的append()
  16. //写入KeyValue信息
  17. StoreFile$Writer#append(){
  18. appendGeneralBloomfilter(kv);
  19. appendDeleteFamilyBloomFilter(kv);
  20. HFileWriterV2.append(kv);
  21. trackTimestamps(kv);
  22. }
  24. //这个方法封装了处理heap取出的记录值的逻辑，
  25. //根据matcher对该值的判断来决定这个值是输出还是跳过
  26. StoreSanner#next(){
  27. KeyValuepeeked=this.heap.peek();
  28. if(peeked==null){
  29. close();
  30. returnfalse;
  31. }
  32. LOOP:
  33. while((kv=this.heap.peek())!=null){
  34. ScanQueryMatcher.MatchCodeqcode=matcher.match(kv);
  35. switch(qcode){
  36. caseINCLUDE:
  37. caseINCLUDE_AND_SEEK_NEXT_ROW:
  38. caseINCLUDE_AND_SEEK_NEXT_COL:
  39. Filterf=matcher.getFilter();
  40. outResult.add(f==nullkv:f.transform(kv));
  41. count++;
  42. if(qcode==ScanQueryMatcher.MatchCode.INCLUDE_AND_SEEK_NEXT_ROW){
  43. if(!matcher.moreRowsMayExistAfter(kv)){
  44. returnfalse;
  45. }
  46. reseek(matcher.getKeyForNextRow(kv));
  47. }elseif(qcode==ScanQueryMatcher.MatchCode.INCLUDE_AND_SEEK_NEXT_COL){
  48. reseek(matcher.getKeyForNextColumn(kv));
  49. }else{
  50. this.heap.next();
  51. }
  52. cumulativeMetric+=kv.getLength();
  53. if(limit>0&&(count==limit)){
  54. breakLOOP;
  55. }
  56. continue;
  57. caseDONE:
  58. returntrue;
  59. caseDONE_SCAN:
  60. close();
  61. returnfalse;
  62. caseSEEK_NEXT_ROW:
  63. if(!matcher.moreRowsMayExistAfter(kv)){
  64. returnfalse;
  65. }
  66. reseek(matcher.getKeyForNextRow(kv));
  67. break;
  68. caseSEEK_NEXT_COL:
  69. reseek(matcher.getKeyForNextColumn(kv));
  70. break;
  71. caseSKIP:
  72. this.heap.next();
  73. break;
  74. caseSEEK_NEXT_USING_HINT:
  75. KeyValuenextKV=matcher.getNextKeyHint(kv);
  76. if(nextKV!=null){
  77. reseek(nextKV);
  78. }else{
  79. heap.next();
  80. }
  81. break;
  82. default:
  83. thrownewRuntimeException("UNEXPECTED");
  84. }//endwhile
  85. }
  87. //KeyValueHeap使用堆排序输出结果
  88. //内部使用了优先队列，再用KVScannerComparator
  89. //作为比较工具
  90. KeyValueHeap#构造函数(){
  91. this.comparator=newKVScannerComparator(comparator);
  92. heap=newPriorityQueue<KeyValueScanner>(scanners.size(),
  93. this.comparator);
  94. for(KeyValueScannerscanner:scanners){
  95. if(scanner.peek()!=null){
  96. this.heap.add(scanner);
  97. }else{
  98. scanner.close();
  99. }
  100. }
  101. this.current=pollRealKV();
  102. }
  104. //堆里面最重要的方法其实就是next，不过看这个方法的主要功能不是
  105. //为了算出nextKeyValue，而主要是为了算出nextScanner，然后需在外部
  106. //再次调用peek方法来取得nextKeyValue，不是很简练。
  107. KeyValueHeap#next(){
  108. InternalScannercurrentAsInternal=(InternalScanner)this.current;
  109. booleanmayContainMoreRows=currentAsInternal.next(result,limit,metric);
  110. KeyValuepee=this.current.peek();
  111. if(pee==null||!mayContainMoreRows){
  112. this.current.close();
  113. }else{
  114. this.heap.add(this.current);
  115. }
  116. this.current=pollRealKV();
  117. return(this.current!=null);
  118. }
  120. //这里省略了其他部分，注意这里有两个赋值
  121. //对于compact来说如果是minor类型的则不会删除掉DELETE类型的KeyValue
  122. //而major类型在最终输出的时候会删除掉DELETE类型的KeyValue标记
  123. ScanQueryMatcher#构造函数(){
  124. //.....
  125. /*howtodealwithdeletes*/
  126. this.isUserScan=scanType==ScanType.USER_SCAN;
  127. this.retainDeletesInOutput=scanType==ScanType.MINOR_COMPACT||scan.isRaw();
  128. //..
  129. }

HRegionServer调用split请求


执行逻辑如下:

  1. //切分region
  2. HRegionServer#splitRegion(){
  3. HRegionregion=getRegion(regionInfo.getRegionName());
  4. region.flushcache();
  5. region.forceSplit(splitPoint);
  6. compactSplitThread.requestSplit(region,region.checkSplit());
  7. }
  9. //创建SplitRequest对象，放到线程池中执行
  10. CompactSplitThread#requestSplit(){
  11. ThreadPoolExecutor#execute(newSplitRequest(r,midKey,HRegionServer.this));
  12. }

split线程执行过程


META表更新的瞬间

主要逻辑如下:

  1. //在单线中执行
  2. SplitRequest#run(){
  3. SplitTransactionst=newSplitTransaction(parent,midKey);
  4. if(!st.prepare()){
  5. return;
  6. }
  7. st.execute(this.server,this.server);
  8. }
  10. //核心逻辑，先创建两个子region，再创建临时的ZK节点
  11. //将父region切分，创建临时目录，将region关闭
  12. //开始切分，将storefile放到目录中
  13. //创建子regionA和B，同时open这两个region，更新META信息
  14. //更新ZK信息，将原region下线
  15. SplitTransaction#execute(){
  16. PairOfSameType<HRegion>regions=createDaughters(server,services);
  17. openDaughters(server,services,regions.getFirst(),regions.getSecond());
  18. transitionZKNode(server,services,regions.getFirst(),regions.getSecond());
  19. }
  22. //预先创建两个子region
  23. SplitTransaction#prepare(){
  24. HRegionInfohri=parent.getRegionInfo();
  25. hri_a=newHRegionInfo(hri.getTableName(),startKey,splitrow,false,rid);
  26. hri_b=newHRegionInfo(hri.getTableName(),splitrow,endKey,false,rid);
  27. }
  29. SplitTransaction#createDaughters(){
  30. //创建类似/hbase/unassigned/fad11edf1e6e0a842b7fd3ad87f25053
  31. //这样的节点，其中的编码后的region就是待split的region
  32. createNodeSplitting();
  33. //用于记录事务的处理进展
  34. this.journal.add(JournalEntry.SET_SPLITTING_IN_ZK);
  36. //将这个节点作为事务节点，待任务处理完后会删除这个节点
  37. transitionNodeSplitting();
  39. //创建类似/hbase/kvdb/fad11edf1e6e0a842b7fd3ad87f25053/.splits
  40. //的HDFS节点，用于临时处理split文件
  41. createSplitDir();
  43. //关闭待处理的region
  44. List<StoreFile>hstoreFilesToSplit=this.parent.close(false);
  45. HRegionServer.removeFromOnlineRegions(this.parent.getRegionInfo().getEncodedName());
  46. splitStoreFiles(this.splitdir,hstoreFilesToSplit);
  47. this.journal.add(JournalEntry.STARTED_REGION_A_CREATION);
  48. HRegiona=createDaughterRegion(this.hri_a,this.parent.rsServices);
  49. this.journal.add(JournalEntry.STARTED_REGION_B_CREATION);
  50. HRegionb=createDaughterRegion(this.hri_b,this.parent.rsServices);
  52. //更新META表信息
  53. MetaEditor.offlineParentInMeta(server.getCatalogTracker(),
  54. this.parent.getRegionInfo(),a.getRegionInfo(),b.getRegionInfo());
  56. //返回两个子regionA和B
  57. returnnewPairOfSameType<HRegion>(a,b);
  58. }
  60. SplitTransaction#splitStoreFiles(){
  61. for(StoreFilesf:hstoreFilesToSplit){
  62. //splitStoreFile(sf,splitdir);
  63. StoreFileSplittersfs=newStoreFileSplitter(sf,splitdir);
  64. futures.add(threadPool.submit(sfs));
  65. }
  66. //等待线程池中的任务执行完后返回
  67. }
  69. //开始分割文件
  70. SplitTransaction$StoreFileSplitter#call(){
  71. splitStoreFile(sf,splitdir);
  72. }
  74. SplitTransaction#splitStoreFile(){
  75. FileSystemfs=this.parent.getFilesystem();
  76. byte[]family=sf.getFamily();
  77. Stringencoded=this.hri_a.getEncodedName();
  78. //地址类似
  79. ///hbase/kvdb/fad11edf1e6e0a842b7fd3ad87f25053/.splits/1977310abc183fac9aba3dc626b01a2d
  80. ///value/92e897822d804d3bb4805548e9a80bd2.fad11edf1e6e0a842b7fd3ad87f25053
  81. Pathstoredir=Store.getStoreHomedir(splitdir,encoded,family);
  82. //这里会根据splitRow分别创建两个文件，一个是从最开始到splitRow
  83. //还有一个是从splitRow到文件最后
  84. //这里是直接调用HDFS的API写入到底层文件系统中的
  85. StoreFile.split(fs,storedir,sf,this.splitrow,Range.bottom);
  86. encoded=this.hri_b.getEncodedName();
  87. storedir=Store.getStoreHomedir(splitdir,encoded,family);
  88. StoreFile.split(fs,storedir,sf,this.splitrow,Range.top);
  89. }
  91. //这里会根据传入的参数，是从开始到splitRow
  92. //还是从splitRow到文件结束
  93. //如果是从开始到splitRow，那么判断第一个key如果splitRow大则这个
  94. //文件就不需要分割了，直接返回即可
  95. StoreFile#split(){
  96. if(range==Reference.Range.bottom){
  97. KeyValuesplitKey=KeyValue.createLastOnRow(splitRow);
  98. byte[]firstKey=f.createReader().getFirstKey();
  99. if(f.getReader().getComparator().compare(splitKey.getBuffer(),
  100. splitKey.getKeyOffset(),splitKey.getKeyLength(),
  101. firstKey,0,firstKey.length)<0){
  102. returnnull;
  103. }
  104. }else{
  105. KeyValuesplitKey=KeyValue.createFirstOnRow(splitRow);
  106. byte[]lastKey=f.createReader().getLastKey();
  107. if(f.getReader().getComparator().compare(splitKey.getBuffer(),
  108. splitKey.getKeyOffset(),splitKey.getKeyLength(),
  109. lastKey,0,lastKey.length)>0){
  110. returnnull;
  111. }
  112. }
  113. Referencer=newReference(splitRow,range);
  114. StringparentRegionName=f.getPath().getParent().getParent().getName();
  115. Pathp=newPath(splitDir,f.getPath().getName()+"."+parentRegionName);
  116. returnr.write(fs,p);
  117. }
  119. //创建一个HRegion
  120. SplitTransaction#createDaughterRegion(){
  121. FileSystemfs=this.parent.getFilesystem();
  122. PathregionDir=getSplitDirForDaughter(this.parent.getFilesystem(),
  123. this.splitdir,hri);
  124. HRegionr=HRegion.newHRegion(this.parent.getTableDir(),
  125. this.parent.getLog(),fs,this.parent.getBaseConf(),
  126. hri,this.parent.getTableDesc(),rsServices);
  127. longhalfParentReadRequestCount=this.parent.getReadRequestsCount()/2;
  128. r.readRequestsCount.set(halfParentReadRequestCount);
  129. r.setOpMetricsReadRequestCount(halfParentReadRequestCount);
  130. longhalfParentWriteRequest=this.parent.getWriteRequestsCount()/2;
  131. r.writeRequestsCount.set(halfParentWriteRequest);
  132. r.setOpMetricsWriteRequestCount(halfParentWriteRequest);
  133. HRegion.moveInitialFilesIntoPlace(fs,regionDir,r.getRegionDir());
  134. returnr;
  135. }
  137. //设置region的info:regioninfo列为下线状态
  138. //再增加两个列info:splitA和info:splitB
  139. MetaEditor#offlineParentInMeta(){
  140. HRegionInfocopyOfParent=newHRegionInfo(parent);
  141. copyOfParent.setOffline(true);
  142. copyOfParent.setSplit(true);
  143. Putput=newPut(copyOfParent.getRegionName());
  144. addRegionInfo(put,copyOfParent);
  145. put.add("info","splitA",Writables.getBytes(a));
  146. put.add("info","splitB",Writables.getBytes(b));
  147. putToMetaTable(catalogTracker,put);
  148. }
  152. //这里的DaughterOpener是对HRegion的封装
  153. //会在新线程中启动HRegion#open()
  154. //之后会更新META表信息，之后META表在很短的时间内
  155. //会同时存在父region信息(已下线)和两个子region信息
  156. SplitTransaction#openDaughters(){
  157. DaughterOpeneraOpener=newDaughterOpener(server,a);
  158. DaughterOpenerbOpener=newDaughterOpener(server,b);
  159. aOpener.start();
  160. bOpener.start();
  161. aOpener.join();
  162. bOpener.join();
  164. HRegionServer.postOpenDeployTasks(b,server.getCatalogTracker(),true);
  165. //ShouldaddittoOnlineRegions
  166. HRegionServer.addToOnlineRegions(b);
  167. HRegionServer.postOpenDeployTasks(a,server.getCatalogTracker(),true);
  168. HRegionServer.addToOnlineRegions(a);
  169. }
  171. //如果StoreFile超过一定数量了会执行compact
  172. //然后更新ZK或者ROOT和META表
  173. HRegionServer#postOpenDeployTasks(){
  174. for(Stores:r.getStores().values()){
  175. if(s.hasReferences()||s.needsCompaction()){
  176. getCompactionRequester().requestCompaction(r,s,"OpeningRegion",null);
  177. }
  178. }
  179. //更新ZK或者ROOT和META表
  180. if(r.getRegionInfo().isRootRegion()){
  181. RootLocationEditor.setRootLocation(getZooKeeper(),
  182. this.serverNameFromMasterPOV);
  183. }elseif(r.getRegionInfo().isMetaRegion()){
  184. MetaEditor.updateMetaLocation(ct,r.getRegionInfo(),
  185. this.serverNameFromMasterPOV);
  186. }else{
  187. if(daughter){
  188. //Ifdaughterofasplit,updatewholerow,notjustlocation.
  189. MetaEditor.addDaughter(ct,r.getRegionInfo(),
  190. this.serverNameFromMasterPOV);
  191. }else{
  192. MetaEditor.updateRegionLocation(ct,r.getRegionInfo(),
  193. this.serverNameFromMasterPOV);
  194. }
  195. }
  196. }
  198. //将ZK中/hbase/unassigned节点下的
  199. //fad11edf1e6e0a842b7fd3ad87f25053(待处理的region)
  200. //删除
  201. SplitTransaction#transitionZKNode(){
  202. transitionNodeSplit();
  203. tickleNodeSplit();
  204. }

一些辅助逻辑:

  1. //等待压缩完成，然后刷新数据
  2. //最后再线程池中关闭所有的Store
  3. HRegion#close(){
  4. waitForFlushesAndCompactions();
  5. internalFlushcache();
  6. ThreadPoolExecutorstoreCloserThreadPool=
  7. getStoreOpenAndCloseThreadPool("StoreCloserThread-"
  8. +this.regionInfo.getRegionNameAsString());
  9. CompletionService<ImmutableList<StoreFile>>completionService=
  10. newExecutorCompletionService<ImmutableList<StoreFile>>(
  11. storeCloserThreadPool);
  13. for(finalStorestore:stores.values()){
  14. completionService.submit(newCallable<ImmutableList<StoreFile>>(){
  15. publicImmutableList<StoreFile>call()throwsIOException{
  16. returnstore.close();
  17. }
  18. });
  19. }
  20. }
  22. //提交到线程池中关闭所有打开的StoreFile
  23. Store#close(){
  24. for(finalStoreFilef:result){
  25. completionService.submit(newCallable<Void>(){
  26. publicVoidcall()throwsIOException{
  27. f.closeReader(true);
  28. returnnull;
  29. }
  30. }
  31. }

compactionChecker线程

这个类是用于定期检查region server下的region是否需要做compact

主要逻辑如下:

  1. //不停的遍历当前RegionServer下的所有Region
  2. //然后检查是否需要做compact
  3. CompactionChecker#chore(){
  4. for(HRegionr:this.instance.onlineRegions.values()){
  5. for(Stores:r.getStores().values()){
  6. if(s.needsCompaction()){
  7. //Queueacompaction.Willrecognizeifmajorisneeded.
  8. this.instance.compactSplitThread.requestCompaction(r,s,getName());
  9. }elseif(s.isMajorCompaction()){
  10. if(majorCompactPriority==DEFAULT_PRIORITY
  11. ||majorCompactPriority>r.getCompactPriority()){
  12. this.instance.compactSplitThread.requestCompaction(r,s,getName());
  13. }else{
  14. this.instance.compactSplitThread.requestCompaction(r,s,getName());
  15. }
  16. }
  17. }
  18. }
  19. }

参考

深入分析HBase Compaction机制

Hbase的Region Compact算法实现分析

深入分析HBase RPC(Protobuf)实现机制

HBase region split源码分析
【大 中 小】【打印】 【繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部】
上一篇：关于hbase应用的一些看法 下一篇：Hbase的存储原理

最新文章

愚公移山  
Hbase架构   Hbase Region的拆..
5 hbase-shell +   hbase的j..
Hbase   MapReduce例子
愚公移山  

热门文章

GeoMesa-HBase部署实践
Oracle GoldenGate 同步数据至Hbase
Flink读写系列之-读HBase并写入HBase
传统的行存储和（HBase）列存储的区..
flink实战--读写Hbase

Hot 文章

GeoMesa-HBase部署实践
Oracle GoldenGate 同步数据至Hbase
Flink读写系列之-读HBase并写入HBase
传统的行存储和（HBase）列存储的区..
flink实战--读写Hbase

Python

C 语言

C++基础

大数据基础

linux编程基础

C/C++面试题目