Procedure V2, 是hbase1.1版本引入的一套fault-tolerant的执行multi-steps-job的框架, 目前主要用在Master中, 比如创建表,删除表等操作
新旧比较
下面比较0.94版本和1.25版本下的建表流程
0.94
0.94版本中,创建表是通过HBaseAdmin类,向Master发起一个异步的建表请求,然后不断的扫描meta表,直到从meta表中扫描到的表的region数目和预期的数据相同,即认为建表成功, 时序图如下:
流程:
1. 从HBaseAdmin发起 createTable请求到master[异步请求]
2. Master做完基本检查后,如果有MasterCoprocessor,就调用coprocessor的preCreateTable做前置检查
3. 检查通过后,Master生成CreateTableHandler, 提交给ExecuterService服务去执行
3.1. hdfs的 hbaseroot/.tmp目录上创建TableDescriptor
3.2. hdfs的hbaseroot/.tmp目录上创建region及相关文件
3.3. 把tmp目录中创建好的表文件拷贝到hbaseroot目录下
3.4. 新增region信息到META表
3.5. 触发所有region的assignment,随机分配到集群alive的regionserver上
3. 6. zk上修改表的状态为TableState.ENABLED
4. HBaseAdmin在第1步完成后, 会通过MetaScan不断的去meta表查看所创建表的region数目是不是和预期的相等,如果是,建表成功,如果经过了一定的重试次数后(默认10*10=100次),依然失败,则抛出TableNotEnabledException异常
存在的问题:
1. 如果一切顺利还好,但是如果任意一步出现问题,则很难定位和修复,比如:
1. 3.3成功后,master宕机, 3.4还没有执行, 那么定位后,可以通过hbase hbck -fixMeta 工具做修复
2. 如果3.4部分成功, 就没有很好的工具做修复, 这时候就需要人工介入,去做回滚操作,然后再重新建表
1.2.5
procedure v2是在hbase1.1版本引入的特性,主要是为了解决之前的方案中存在的问题, 比如,任务意外中断后,中间状态需要人工接入, 难以追溯任务状态等问题, 用户可以选择是否开启procedure特性,如果不开启,就还是使用类似0.94版本的Handler机制.
Procedure有以下核心组件:
其中, ProcedureExecuter负责procedure的编排和运行, procedureStore用于持久化procedure的状态
Procedure 有以下状态:
每个Procedure都会经过这几个状态,每次状态的变化,都会通过ProcedureStore记录到WAL中
Procedure V2有以下几个核心接口类:
1. Procedure 代表一个任务的基础类, 主要包含两个接口方法:
a. execute
b. rollback
2. ProcedureStore 持久化Procedure状态的类, 机器宕机恢复后会通过ProcedureStore获取之前没有运行完的任务状态,并且继续执行
3. ProcedureExecutor 类似于ExecutorService, 用于执行procedure,同时也负责Procedure任务的恢复和rollback等操作, 复杂程度超过ExecutorService
使用Procedure v2的场景下,时序图如下:
通过Procedure V2,可以将一个任务拆分成多个子Procedure, 具体的实现方式是, 父任务的execute返回多个Procedure子任务,这个时候, 父任务的状态变成Waiting, 并继续完成子任务,当所有子任务都成功后, 父任务继续执行
根据这个特性, hbase实现了一个StateMachineProcedure, 基于状态机, CreateTableProcedure被分成了多个状态:
enum CreateTableState {
CREATE_TABLE_PRE_OPERATION = 1;
CREATE_TABLE_WRITE_FS_LAYOUT = 2;
CREATE_TABLE_ADD_TO_META = 3;
CREATE_TABLE_ASSIGN_REGIONS = 4;
CREATE_TABLE_UPDATE_DESC_CACHE = 5;
CREATE_TABLE_POST_OPERATION = 6;
}
初始化状态下, 状态为CREATE_TABLE_PRE_OPERATION, 状态变化逻辑在ExecuteFromState中实现:
@Override
protected Flow executeFromState(final MasterProcedureEnv env, final CreateTableState state)
throws InterruptedException {
if (LOG.isTraceEnabled()) {
LOG.trace(this + " execute state=" + state);
}
try {
switch (state) {
case CREATE_TABLE_PRE_OPERATION:
// Verify if we can create the table
boolean exists = !prepareCreate(env);
ProcedurePrepareLatch.releaseLatch(syncLatch, this);
if (exists) {
assert isFailed() : "the delete should have an exception here";
return Flow.NO_MORE_STATE;
}
preCreate(env);
setNextState(CreateTableState.CREATE_TABLE_WRITE_FS_LAYOUT);
break;
case CREATE_TABLE_WRITE_FS_LAYOUT:
newRegions = createFsLayout(env, hTableDescriptor, newRegions);
setNextState(CreateTableState.CREATE_TABLE_ADD_TO_META);
break;
case CREATE_TABLE_ADD_TO_META:
newRegions = addTableToMeta(env, hTableDescriptor, newRegions);
setNextState(CreateTableState.CREATE_TABLE_ASSIGN_REGIONS);
break;
case CREATE_TABLE_ASSIGN_REGIONS:
assignRegions(env, getTableName(), newRegions);
setNextState(CreateTableState.CREATE_TABLE_UPDATE_DESC_CACHE);
break;
case CREATE_TABLE_UPDATE_DESC_CACHE:
updateTableDescCache(env, getTableName());
setNextState(CreateTableState.CREATE_TABLE_POST_OPERATION);
break;
case CREATE_TABLE_POST_OPERATION:
postCreate(env);
return Flow.NO_MORE_STATE;
default:
throw new UnsupportedOperationException("unhandled state=" + state);
}
} catch (HBaseException|IOException e) {
LOG.error("Error trying to create table=" + getTableName() + " state=" + state, e);
setFailure("master-create-table", e);
}
return Flow.HAS_MORE_STATE;
}
如果CreateTable过程中,在某一个状态下运行失败需要回滚, ProcedureExecutor会负责追溯回滚procedure的父Procedure,直到全部回滚; 如果Procedure执行或者回滚过程中出现宕机,Master在启动的时候,会启动ProcedureExecutor,从WAL中恢复出之前Procedure的状态;
ProcedureStore的实现
...未完待续