region split流程分析
region split流程分析
region split流程分析
splitregion的发起主要通过client端调用regionserver.splitRegion或memstore.flsuh时检查并发起。
Client通过rpc调用regionserver的splitRegion方法
client端通过HBaseAdmin.split传入regionname与splitpoint(切分的rowkey,可以不传入),
通过meta得到此region所在的server，发起rpc请求，调用HRegionServer.splitRegion方法
publicSplitRegionResponse splitRegion(finalRpcController controller,
finalSplitRegionRequest request)throwsServiceException {
try{
checkOpen();
requestCount.increment();
从onlineRegions中拿到对应的HRegion
HRegion region= getRegion(request.getRegion());
region.startRegionOperation(Operation.SPLIT_REGION);
LOG.info("Splitting" +region.getRegionNameAsString());
在做split前，先对region进行flush操作。请参见regionflush流程分析
region.flushcache();
如果client端发起split请求时指定有split的rowkey，拿到splitkey的值
byte[]splitPoint= null;
if(request.hasSplitPoint()){
splitPoint= request.getSplitPoint().toByteArray();
}
设置region的splitRequest属性为true,表示有splitrequest
如果splitrowkey传入不为空，也就是指定有rowkey，设置region的explicitSplitPoint为指定的rowkey
但此值设置后不会被清空，原因后面分析
region.forceSplit(splitPoint);
发起splitrequest,到split的线程进行处理。
Region.checkSplit流程：
a.检查region是否是meta/namespace的region,如果是返回null
b.如果hbase.master.distributed.log.replay配置为true时，同时openRegion后此region还没有被replay
region.isRecovering==true,如果是返回null
c.通过hbase.regionserver.region.split.policy配置的RegionSplitPolicy,
默认为IncreasingToUpperBoundRegionSplitPolicy,
也可以直接在tablecreate时配置SPLIT_POLICY的值为class
调用splitPolicy.shouldSplit(),此方法做如下检查流程，并返回true与false
c.1检查region.splitRequest的值是否为true,如果是，返回true,
c.2得到当前regionserver中此region对应的table的所有region个数，
得到region的存储最大size,
c.2.1取出通过tablecreate时的属性MAX_FILESIZE来设置的region最大存储大小，
如果没有取hbase.hregion.max.filesize配置的的值，默认为10* 1024 * 1024 * 1024L(10g)
c.2.2取出通过tablecreate时的MEMSTORE_FLUSHSIZE属性来设置的regionmemstore的大小，
如果没有取hbase.hregion.memstore.flush.size配置的值，默认为1024*1024*128L(128M)
c.2.3通过c.2.2的值*(当前rs中此table的region个数*当前rs中此table的region个数)
c.2.4取出c.2.1中得到的值与c.2.3计算出的值最小的一个，得到region最大可存储的size值
c.3检查region中所有store中是否有reference的storefile，如果有返回false
c.4检查region中所有store中所有的storefile的大小是否超过了c.2中得到的size大小，如果是返回true
d.如果c方法调用返回的结果是false,返回null
f.调用RegionSplitPolicy.getSplitPoint()方法返回进行split的切分rowkey
f.1如果region.explicitSplitPoint的值不为空，返回此值
f.2迭代region中所有的store,调用HStore.getSplitPoint()方法得到此store的splitrowkey
HStore.getSplitPoint()方法流程：
调用this.storeEngine.getStoreFileManager().getSplitPoint();得到一个splitpoint
通过hbase.hstore.engine.class配置storeEngine的实现类，默认为DefaultStoreEngine
默认的storeFileManager为DefaultStoreFileManager，如果store中没有storefile，返回null
否则得到size最大的storefile,并得到此storefile的中间rowkey，并返回此值
g.检查f中得到的rowkey是否在region中，如果不在返回null,否则返回此rowkey,到此checkSplit流程完成
requestSplit见下面的compactSplitThread.requestSplit(region,rowkey)流程分析
compactSplitThread.requestSplit(region,region.checkSplit());
returnSplitRegionResponse.newBuilder().build();
}catch(IOException ie){
thrownewServiceException(ie);
}
}
compactSplitThread.requestSplit(region,rowkey)流程
此方法是所有的split请求的最终执行处理程序
publicsynchronized voidrequestSplit(finalHRegion r, byte[]midKey) {
如果进行split操作的传入进行切分region的rowkey为null,不做split操作
if(midKey ==null){
LOG.debug("Region" + r.getRegionNameAsString()+
"not splittable because midkey=null");
return;
}
try{
生成一个SplitRequest执行线程，通过splits线程池执行此线程，
线程池大小通过hbase.regionserver.thread.split配置，默认为1
this.splits.execute(newSplitRequest(r,midKey,this.server));
if(LOG.isDebugEnabled()){
LOG.debug("Splitrequested for " + r+ ". "+ this);
}
}catch(RejectedExecutionException ree){
LOG.info("Couldnot execute split for " + r,ree);
}
}
SplitRequest.run方法处理流程：
publicvoid run(){
if(this.server.isStopping()|| this.server.isStopped()){
LOG.debug("Skippingsplit because server is stopping="+
this.server.isStopping()+ " or stopped="+ this.server.isStopped());
return;
}
try{
finallongstartTime =System.currentTimeMillis();
生成一个split执行程序
SplitTransaction st= newSplitTransaction(parent,midKey);
//acquirea shared read lock on the table, so that table schema modifications
//donot happen concurrently
tableLock=server.getTableLockManager().readLock(parent.getTableDesc().getTableName()
, "SPLIT_REGION:"+ parent.getRegionNameAsString());
try{
tableLock.acquire();
} catch(IOException ex){
tableLock= null;
throwex;
}
//If prepare does not return true, for some reason -- logged inside in
//the prepare call -- we are not ready to split just now. Just return.
根据splitkey把原来的region的startkey到splitkey与currenttime生成一个regioninfo
根据splitkey把原来的region的splitkey到endkey与currenttime生成一个regioninfo
if(!st.prepare())return;
try{
执行split操作，通过hbase.regionserver.fileSplitTimeout配置splitfile的timeout时间，默认为30000ms
在zk中的region-in-transition路径下生成一个根据此region的子路径的RegionTransition实例，
此实例在zk上存储的值为新生成的两个hregioninfo信息，
并设置zk中此节点的EventType为RS_ZK_REQUEST_REGION_SPLIT
在hdfs中的此region目录下生成一个.splits目录,
关闭当前的region,并得到当前region中所有的store与store下的storefile列表。
从rs中的onlineRegions列表中移出此region
迭代每一个store中的所有storefile，生成一个SplitTransaction.StoreFileSplitter实例
通过HRegionFileSystem.splitStoreFile生成一个以splitrow结束的与一个以splitrow开头的Reference的hfile
并存储在切分后的两个新的region的.splits/cfname/storefilename.oldregionname文件
通过调用HRegion(oldregion).createDaughterRegionFromSplits(newregionInfo)生成两个新的HRegion实例
并把.splits目录下的hfile文件move到newregion的目录下
更新meta表中的信息
生成SplitTransaction.DaughterOpener线程实例，在当前rs中直接通过openregion打开两个新的hregion实例。
把zk中节点的transition的zkEventType从RS_ZK_REGION_SPLITTING更新到RS_ZK_REGION_SPLIT，
通知master在regionserver中的split完成，等待master对这个消息进行处理。直到master处理完成。
st.execute(this.server,this.server);
} catch(Exception e){
if(this.server.isStopping()|| this.server.isStopped()){
LOG.info(
"Skiprollback/cleanup of failed split of "
+parent.getRegionNameAsString()+ " because server is"
+(this.server.isStopping()? " stopping": " stopped"),e);
return;
}
try{
LOG.info("Runningrollback/cleanup of failed split of "+
parent.getRegionNameAsString()+ "; "+ e.getMessage(),e);
if(st.rollback(this.server,this.server)){
LOG.info("Successfulrollback of failed split of " +
parent.getRegionNameAsString());
} else{
this.server.abort("Abort;we got an error after point-of-no-return");
}
} catch(RuntimeException ee){
String msg= "Failed rollback of failed splitof " +
parent.getRegionNameAsString()+ " -- aborting server";
//If failed rollback,kill this server to avoid having a hole in table.
LOG.info(msg,ee);
this.server.abort(msg);
}
return;
}
LOG.info("Regionsplit, hbase:meta updated, and report to master. Parent="
+parent.getRegionNameAsString()+ ", new regions: "
+st.getFirstDaughter().getRegionNameAsString()+ ", "
+st.getSecondDaughter().getRegionNameAsString()+ ". Split took "
+StringUtils.formatTimeDiff(System.currentTimeMillis(),startTime));
}catch(IOException ex){
LOG.error("Splitfailed " + this,RemoteExceptionHandler.checkIOException(ex));
server.checkFileSystem();
}finally{
if(this.parent.getCoprocessorHost()!= null){
try{
this.parent.getCoprocessorHost().postCompleteSplit();
} catch(IOException io){
LOG.error("Splitfailed " + this,
RemoteExceptionHandler.checkIOException(io));
}
}
releaseTableLock();
}
}
master中处理regionsplit的监听流程
通过AssignmentManager.nodeDataChange事件监听rs中对splitregion的值修改。
nodeDataChanged-->handleAssignmentEvent-->handleRegion
switch(rt.getEventType()){
caseRS_ZK_REQUEST_REGION_SPLIT:
caseRS_ZK_REGION_SPLITTING:
caseRS_ZK_REGION_SPLIT:
设置两个新的region的状态为online状态。并删除zk上的路径
if(!handleRegionSplitting(
rt,encodedName,prettyPrintedRegionName,sn)) {
deleteSplittingNode(encodedName,sn);
}
break;
执行memstore的flush后的split流程分析
在每次执行完成memstore时，会进行是否需要split的检查，如果需要进行split，会发起splitrequest操作。
privatebooleanflushRegion(finalHRegion region,finalbooleanemergencyFlush){
...............................................此处省去一些代码
Region.checkSplit流程：
a.检查region是否是meta/namespace的region,如果是返回null
b.如果hbase.master.distributed.log.replay配置为true时，同时openRegion后此region还没有被replay
region.isRecovering==true,如果是返回null
c.通过hbase.regionserver.region.split.policy配置的RegionSplitPolicy,
默认为IncreasingToUpperBoundRegionSplitPolicy,
也可以直接在tablecreate时配置SPLIT_POLICY的值为class
调用splitPolicy.shouldSplit(),此方法做如下检查流程，并返回true与false
c.1检查region.splitRequest的值是否为true,如果是，返回true,
c.2得到当前regionserver中此region对应的table的所有region个数，
得到region的存储最大size,
c.2.1取出通过tablecreate时的属性MAX_FILESIZE来设置的region最大存储大小，
如果没有取hbase.hregion.max.filesize配置的的值，默认为10* 1024 * 1024 * 1024L(10g)
c.2.2取出通过tablecreate时的MEMSTORE_FLUSHSIZE属性来设置的regionmemstore的大小，
如果没有取hbase.hregion.memstore.flush.size配置的值，默认为1024*1024*128L(128M)
c.2.3通过c.2.2的值*(当前rs中此table的region个数*当前rs中此table的region个数)
c.2.4取出c.2.1中得到的值与c.2.3计算出的值最小的一个，得到region最大可存储的size值
c.3检查region中所有store中是否有reference的storefile，如果有返回false
c.4检查region中所有store中所有的storefile的大小是否超过了c.2中得到的size大小，如果是返回true
d.如果c方法调用返回的结果是false,返回null
f.调用RegionSplitPolicy.getSplitPoint()方法返回进行split的切分rowkey
f.1如果region.explicitSplitPoint的值不为空，返回此值
f.2迭代region中所有的store,调用HStore.getSplitPoint()方法得到此store的splitrowkey
HStore.getSplitPoint()方法流程：
调用this.storeEngine.getStoreFileManager().getSplitPoint();得到一个splitpoint
通过hbase.hstore.engine.class配置storeEngine的实现类，默认为DefaultStoreEngine
默认的storeFileManager为DefaultStoreFileManager，如果store中没有storefile，返回null
否则得到size最大的storefile,并得到此storefile的中间rowkey，并返回此值
g.检查f中得到的rowkey是否在region中，如果不在返回null,否则返回此rowkey,到此checkSplit流程完成
此处主要是检查region中是否有store的大小超过了配置的指定大小，也就是对c的检查
booleanshouldSplit= region.checkSplit()!= null;
if(shouldSplit){
如果需要做split操作，发起splitrequest
this.server.compactSplitThread.requestSplit(region);
} elseif(shouldCompact){
server.compactSplitThread.requestSystemCompaction(
region,Thread.currentThread().getName());
}
...............................................此处省去一些代码
returntrue;
}
publicsynchronized booleanrequestSplit(finalHRegion r) {
//don't split regions that are blocking
a.检查hbase.regionserver.regionSplitLimit配置的splitlimit是否大于rs中的onlineRegions的个数
如果不想做split操作，可以把此值设置为一个较小的值，比如1
b.迭代region下的所有store,检查hbase.hstore.blockingStoreFiles配置的store的文件个数,默认为7
减去store中所有的storefile的个是是否大于或等于Store.PRIORITY_USER(1)
if(shouldSplitRegion()&& r.getCompactPriority()>= Store.PRIORITY_USER){
如果需要做split操作，得到split的key,此时默认从最大的storefile的中间key开始split
byte[]midKey =r.checkSplit();
if(midKey !=null){
发起splitrequest,见compactSplitThread.requestSplit(region,rowkey)流程
requestSplit(r,midKey);
returntrue;
}
}
returnfalse;
}
评论暂时关闭