开发者俱乐部

标题: hbase的Region分裂代码分析（代立冬） [打印本页]

---

作者: jack    时间: 2016-3-3 02:39
标题: hbase的Region分裂代码分析（代立冬）
region分裂有2种触发情景：1是用户手动触发(参见HRegionServer的splitRegion方法)，2是后台flush线程flush完一个region的memstore时，会去检查这个region是否需要分裂(参见MemStoreFlushe的flushRegion方法)。这两种情景在代码实现上并无多大差异。
1.下面以手动的split为例分析，手动split有HregionServer的splitRegion开始

• @Override//手动split的实现  
• public void splitRegion(HRegionInfo regionInfo, byte[] splitPoint)  
•     throws NotServingRegionException, IOException {  
•   checkOpen();  
•   HRegion region = getRegion(regionInfo.getRegionName());  
•   region.flushcache();//刷新memstore，减少内存堆积  
•   region.forceSplit(splitPoint);//强制split  
•   compactSplitThread.requestSplit(region, region.checkSplit());//通过compactSplitThread线程池完成split，具体由SplitRequest的run方法负责，SplitRequest内部会创建一个SplitTransaction来完成split   
• }  
  

其中compactSplitThread.requestSplit(region, region.checkSplit())中region.checkSplit()会计算该region的分裂点,看代码

• public byte[] checkSplit() {  
•   // Can't split ROOT/META  
•   //默认使用IncreasingToUpperBoundRegionSplitPolicy的分裂检查实现，有两种情况需要分裂：  
•   //        1.splitRequest=true  
•   //        2.check到该region下有store大于阀值，这个阀值和hbase.hregion.max.filesize和该region所在的RegionServer上和该Region属于同一表的region个数有关，具体见 getSizeToCheck方法。注意，如果该region下存在一个storefile是reference类型的文件则不能split  
•   if (!splitPolicy.shouldSplit()) {
•     return null;  
•   }  
•    
•   //判定该region是否分裂，如果有reference的storefile则不分裂，否则使用StoreFile.Reader得到最大的storefile，通过HFileBlockIndex得到该最大storefile的midkey   TODO midkey的获得还需深入细看
•   byte[] ret = splitPolicy.getSplitPoint();  
•   
•   return ret;  
• }  
  

2.接下来接着看SplitRequest的run方法主要部分

• SplitTransaction st = new SplitTransaction(parent, midKey);  
• if (!st.prepare()) return;  //再次判断有没有reference的region，创建两个新的region对象，代表分裂后产生的两个dautghter region  
• st.execute(this.server, this.server);
  

下面分析下SplitRequest的execute做了什么

• public PairOfSameType<HRegion> execute(final Server server,   
•     final RegionServerServices services)   
• throws IOException {   
•   PairOfSameType<HRegion> regions = createDaughters(server, services);//在zk上创建一个ephemeral node，以防regionserver在分裂过程中down掉，Zookeeper临时路径是/hbase/region-in-transition/regionEncodedName，在parent region的hdfs下创建.splits文件夹，关闭当前待分裂region  
•   openDaughters(server, services, regions.getFirst(), regions.getSecond());   
•   transitionZKNode(server, services, regions.getFirst(), regions.getSecond());   
•   return regions;   
• }
  

先分析createDaughters

• PairOfSameType<HRegion> createDaughters(final Server server,   
•       final RegionServerServices services) throws IOException {   
•    this.fileSplitTimeout = testing ? this.fileSplitTimeout :  
•       server.getConfiguration().getLong("hbase.regionserver.fileSplitTimeout",  
•           this.fileSplitTimeout);//split超时时间，默认30s  
•    if (server != null && server.getZooKeeper() != null) {  
•       try {  
•         createNodeSplitting(server.getZooKeeper(),  
•           this.parent.getRegionInfo(), server.getServerName());//在zk创建一个临时的节点，保存split状态为RS_ZK_REGION_SPLITTING，表示开始region分裂  
•       } catch (KeeperException e) {  
•         throw new IOException("Failed creating SPLITTING znode on " +  
•           this.parent.getRegionNameAsString(), e);  
•       }  
•    }  
•     createSplitDir(this.parent.getFilesystem(), this.splitdir);  //在hdfs上为这个region的split过程创建临时工作目录/hbase/tableName/regionEncodedName/.splits  
•     this.journal.add(JournalEntry.CREATE_SPLIT_DIR);   
•      
•     List<StoreFile> hstoreFilesToSplit = null;   
•     Exception exceptionToThrow = null;   
•     try{   
•       hstoreFilesToSplit = this.parent.close(false);  //关闭当前region，关闭前会等待region的flush和compact都完成(通过writestate同步实现)，还会判断memstore的size小于5m(默认)时，会preFlush,然后关闭该region,region停止读写  
•     } catch (Exception e) {   
•       exceptionToThrow = e;   
•     }   
•      
•     if (!testing) {   
•       services.removeFromOnlineRegions(this.parent.getRegionInfo().getEncodedName());  //从regionserver的online服务中移除  
•     }   
•     this.journal.add(JournalEntry.OFFLINED_PARENT);   
•    
•     splitStoreFiles(this.splitdir, hstoreFilesToSplit);//通过创建与该region下storefile个数相同的线程池子进行并行分裂,见StoreFileSplitter的splitStoreFile方法,其核心走StoreFile.split方法  
•    
•     this.journal.add(JournalEntry.STARTED_REGION_A_CREATION);   
•     HRegion a = createDaughterRegion(this.hri_a, this.parent.rsServices); //region读写数为父region的一半  
•     this.journal.add(JournalEntry.STARTED_REGION_B_CREATION);   
•     HRegion b = createDaughterRegion(this.hri_b, this.parent.rsServices);   
•    
•     if (!testing) {  
•     //在.META.表中下线split的region，修改.META.表的该region信息，把offline split设置为true，添加列：splitA和splitB  
•         MetaEditor.offlineParentInMeta(server.getCatalogTracker(), this.parent.getRegionInfo(), a.getRegionInfo(), b.getRegionInfo());  
•     }  
• ｝  
  

StoreFile的split方法

• static Path split(final FileSystem fs,   final Path splitDir,   final StoreFile f,  final byte [] splitRow, final Reference.Range range)   
•     throws IOException {   
•       
•   // 检查split的key是否在storefile中  
•   if (range == Reference.Range.bottom) {   
•     //check if smaller than first key   
•     KeyValue splitKey = KeyValue.createLastOnRow(splitRow);   
•     byte[] firstKey = f.createReader().getFirstKey();   
•     // If firstKey is null means storefile is empty.   
•     if (firstKey == null) return null;   
•     if (f.getReader().getComparator().compare(splitKey.getBuffer(),     
•         splitKey.getKeyOffset(), splitKey.getKeyLength(),     
•         firstKey, 0, firstKey.length) < 0) {   
•       return null;   
•     }         
•   }   
•   else {   
•     //check if larger than last key.   
•     KeyValue splitKey = KeyValue.createFirstOnRow(splitRow);   
•     byte[] lastKey = f.createReader().getLastKey();         
•     // If lastKey is null means storefile is empty.   
•     if (lastKey == null) return null;   
•     if (f.getReader().getComparator().compare(splitKey.getBuffer(),     
•         splitKey.getKeyOffset(), splitKey.getKeyLength(),     
•         lastKey, 0, lastKey.length) > 0) {   
•       return null;   
•     }   
•   }   
•   
•   /*生成类型为reference的storefile文件,比如encode name为a，column family为cf(该cf下有名为hfile的storefile)的region分裂后会形成名为b和c的引用文件，此时在hdfs中该region下的目录结构为
• /hbase/tableName/a/cf/hfile
• /hbase/tableName/b/.splits/cf/hfile.a
• /hbase/tableName/c/.splits/cf/hfile.a
• 这两个引用文件的storefile的内容由原storefile的中间rowkey和range组成，reference文件的个数与原split region的storefile文件个数相同
• */  
•   Reference r = new Reference(splitRow, range);   
•   String parentRegionName = f.getPath().getParent().getParent().getName();   
•   Path p = new Path(splitDir, f.getPath().getName() + "." + parentRegionName);   
•   return r.write(fs, p);   
• }   
  

再来看一下openDaughters

• void openDaughters(final Server server,   
•       final RegionServerServices services, HRegion a, HRegion b)   
•       throws IOException {   
•       
•       //并行打开两个daughters  
•       DaughterOpener aOpener = new DaughterOpener(server, a);   
•       DaughterOpener bOpener = new DaughterOpener(server, b);   
•       aOpener.start();   
•       bOpener.start();   
•       
•       if (services != null) {   
•         try {   
•           services.postOpenDeployTasks(b, server.getCatalogTracker(), true); // compact有references的storefile，compact操作最终清理掉这些reference文件，并把实际文件的内容写到region中去。将regioninfo信息和location的位置信息put到.META.表中  
•           services.addToOnlineRegions(b);  //添加region对象到regionserver的online列表中，终于可以对外提供服务了  
•           services.postOpenDeployTasks(a, server.getCatalogTracker(), true);   
•           services.addToOnlineRegions(a);   
•         } catch (KeeperException ke) {   
•           throw new IOException(ke);   
•         }   
•       }   
•   }   
  

最后梳理下整个流程：
检查该region是否需要分裂，如果满足分裂条件，则通过region.checkSplit()拿到midkey，并把该分裂请求SplitRequest提交给后台的CompactSplitThread线程池去执行，SplitRequest内部会创建SplitTransaction来实现分裂，其过程如下：
* 根据该region和midkey创建两个新的region对象HRegionInfo，代表分裂后的两个dautghter region  
* 在zk上创建一个临时节点(名称为“/hbase/region-in-transition/region-name”的znode)，以防regionserver在分裂过程中down掉，保存split状态为RS_ZK_REGION_SPLITTING，表示开始region分裂。同时因为master一直watch znode(/hbase/region-in-transition),所以master会知道这个region的变化，以防master对其进行move等操作
* 在该region所在的hdfs路径下创建.splits文件夹
* 关闭该region，关闭前会等待region的flush和compact都完成(通过writestate同步实现)，还会判断如果memstore的size小于5m(默认)时，会preFlush,然后关闭该region,region停止读写,并从regionserver的online服务中移除
* 通过创建与该region下storefile个数相同的线程池子进行storefile的并行分裂,见StoreFileSplitter的splitStoreFile方法,其核心走StoreFile.split方法，其生成类型为reference的storefile文件,比如encode
name为a，column family为cf(该cf下有名为hfile的storefile)的region分裂后会形成名为b和c的引用文件，此时在hdfs中该region下的目录结构为
/hbase/tableName/a/cf/hfile

/hbase/tableName/b/.splits/cf/hfile.a

/hbase/tableName/c/.splits/cf/hfile.a

这两个引用文件的storefile的内容由原storefile的中间rowkey和range组成，reference文件的个数与原split region的storefile文件个数相同


* 在.META.表中下线split的region，修改.META.表的该region信息，把offline split设置为true，添加列：splitA和splitB  
*  并行打开两个daughters region，CompactSplitThread后台线程会compact有references的storefile，compact操作最终清理掉这些reference文件，把实际文件的内容写到daughters region中去。并将daughter region的regioninfo信息和location的位置信息put到.META.表中
* 添加region对象到regionserver的online列表中，终于可以对外提供服务了



参考文章：
http://blog.csdn.net/c77_cn/article/details/38758545
http://www.cnblogs.com/foxmailed/p/3970050.html

---

欢迎光临 开发者俱乐部 (http://xodn.com/)

Powered by Discuz! X3.2