hbase-高级API-协处理器（Observer）

协处理器

过滤器减少服务器端通过网络返回到客户端 的数据量

协处理器（coprocessor）：HBase把部分计算移到数据的存放端

协处理器简介

可理解为服务端的拦截器，可根据需求确定拦截点，再重写这些拦截点对应的方法

客户端的读取API配合筛选机制可控制返回的数据量

进一步优化，例，数据的处理流程直接放到服务器端执行，然后仅返回一个小的处理结果集。类似于一个小型的MapReduce 框架，该框架将工作分发到整个集群

协处理器允许

在region服务器上运行自己的代码，更准确地说是允许用户执行 region级的操作，并且可以使用与RDBMS中触发器（trigger)类似的功能

扩展现有的RPC协议来引入自己的调用（编特定实现特定接口的Java类。用户将其编译成JAR文件，使服务器端可加载。region服务器进程会实例化这些类，并在正确的系统环境中运行。与过滤器不同的是，协处理器可以动态加载。这一点可使用户在HBase集群运行中扩展其功能），这些调用由客户端触发，并在服务端器执行

协处理器场景

使用钩子关联行修改操作来维护一个辅助索引

维护一些数据间的引用完整性

过滤器也可以被增强为有状态的？，因此它们可以做一些跨行级的决策

RDBMS中常见的sum()、avg()等聚合函数，以及SQL也可以在服务器端完成，服务器端只需在本地扫描并统计数据，然后把数值结果返回给客户端

协处理器框架提供了一些类（observer和 endpoint），可通过继承这些类来扩展自己的功能

observer

这类协处理器与触发器（trigger 特殊的存储过程由事件来触发)类似：回调函数（也称钩子函数，hook) 在一些特定事件发生时被执行。包括一些用户产生的事件，也包括服务器端内部自动产生的事件

RegionObserver：可用这种处理器处理数据修改事件，它们与表的region联系紧密。
MasterObserver：可以被用作管理或DDL类型的操作，这些是集群级事件
WALObserver：提供控制WAL的钩子函数

observer提供了一些设计好的回调函数（钩子），每个操作在集群服务器端都可以被调用

endpoint

自定义操作添加到服务器端。用户代码可被部署到管理数据的服务器端，例如，做一些服务器端计算的工作

endpoint通过添加一些远程过程调用来动态扩展RPC协议。可把它们理解为存储过程。endpoint可以与observer的实现组合起来直接作用于服务器端的状态

这些（类）都基于Coprocessor框架，以获取一些共有的特性，同时也可以实现自己特有的功能

Coprocessor 类（接口）

所有协处理器的类都必须实现这个接口。它定义了协处理器的基本约定，并使得框架的管理变得容易。这里提供了两个被应用于框架的枚举类—— Priority和 State

Coprocessor. Priority枚举类定义的优先级（执行顺序）
SYSTEM      高优先级，定义最先被执行的协处理器
USER        定义其他的协处理器，按顺序执行

在同一个优先级中还有一个序号( sequence number) 的概念，用来维护 协处理器的加载顺序。序号从0 开始依次增加，协处理器按照序号顺序执行

协处理器的生命周期中，它们由框架管理。Coprocessor接口提供了如下两个方法:

void start(CoprocessorEnvironment env) throws IOException;
void stop(CoprocessorEnvironment env) throws IOException;

在协处理器开始和结束时被调用。CoprocessorEnvironment用来在协处理器的生命周期中保持其状态。协处理器实例一直被保存在提供的环境中（start和stop之间）

CoprocessorEnvironment 类提供的方法
String getHBaseVersion()        以字符串的格式返回HBase版本
int getVersion()                返回Coprocessor接口的版本
Coprocessor getlnstance()       返回加载的协处理器实例
Coprocessor.Priority getPriority() int getLoadSequence()    协处理器的序号，当协处理器加载时被设置，这反映了它
的执行顺序
HTablelnterface getTable(byte[] tableName)      返回与传入的表名参数对应的HTable实例，这允许协处理 器访问实际的表数据

协处理器应只与提供给它们的环境进行交互。这样可保证没有会被恶意代码用来破坏数据的后门

协处理器应当使用getTable()方法访问表数据

在协处理器实例的生命周期中，Coprocessor接口的start ()和 stop ()方法会被框架隐式调用，处理过程中的每一步都有一个状态

Coprocessor. State枚举类定义的状态
UNINSTALLED 协处理器最初的状态，没有环境，也没有被初始化
INSTALLED   实例装载了它的环境参数
STARTING    协处理器将要开始工作，也就是说start ()方法将要被调用
ACTIVE      一旦start ()方法被调用，当前状态设置为active
STOPPING    stop()方法被调用之前的状态
STOPPED     一旦stop()方法将控制权交给框架，协处理器会被设置为状态stopped

CoprocessorHost类，维护所有协处理器实例和它们专用的环境

Coprocessor、CoprocessorEnvironment 和 CoprocessorHost 这 3 个类形成了协处理器类的基础，基于这3 个类能够实现更高级的功能。它们支持协处理器的生命周期，管理 协处理器的状态，同时提供了执行时的环境参数，以保证协处理器正确执行

首先执行系统级协处理器，然后是用户级协处理器， 并按它们加载时的顺序执行



结果反馈给HRegion和客户端

协处理器加载

处理器配置为使用静态方式加载，也可在集群运行时动态加载协处理器

配置文件和表模式加载是静态加载方法

从配置中加载

通过添加以下配置中的一条或几条到hbase-site.xml文件中来添加协处理器

<property>
<name>hbase.coprocessor.region.classes</name> <value>coprocessor.RegionObserverExample,coprocessor.AnotherCoprocessor </value>
</property>
<property>
<name>hbase.coprocessor.master.classes</name> <value>coprocessor.MasterObserverExample</value>
</property>
<property>
<name>hbase.coprocessor.wal.classes</name> <value>coprocessor.WALObserverExample,bar.foo.MyWALObserver</value>
</property>
将例子中的类名替换为用户自己的类名

配置文件的顺序就是执行顺序，而且这里的协处理器是以系统级加载的。用户应当将全局类配置在这里

配置文件首先在HBase启动时被检查。在配置文件中配置的协处理器是被最先执行的，这 3 项配置中只有一个会被与之相对应的CoprocessorHost的实现加载。例如 ，hbase.coprocessor.master .classes中定义的协处理器会被 MasterCoprocessorHost 类加载

配置项和它们的作用
hbase.coprocessor.master.classes MasterCoprocessorHost  master服务器 hbase.coprocessor.region.classes RegionCoprocessorHost    region服务器 hbase.coprocessor.wal.classes      WALCoprocessorHost     region服务器

当一张表的region被打开时，hbase. coprocessor .region, classes定义的协处理器会被加载。注意用户不能指定具体是哪张表或哪个region加载这个类：默认的协处理器会被每张表和每个region加载。用户在设计自己的协处理器时要注意这点

从表描述符中加载

决定哪些协处理器被加载的选项是表描述符。因为这是针对特定表的

加 载的协处理器只针对这个表的region,同时也只被这些region的region服务器使用

换句话说，只能在与region相关的协处理器上使用这种方法，而不能在master或 WAL 相关的协处理器上使用

这种方法加载的协处理器具有针对性。用户需要在表描述符中利用 HTableDescriptor.setValue()方法定义它们。键必须以COPROCESSOR开头，值必须符合以下格式：

不同级别的表描述符加载
'COPROCESSORS$1 =〉 \ 'hdfs: "localhost:8020/users/leon/test.jar|coprocessor.Test|SYSTEM'
'COPROCESSOR$2' => \
'/Users/laura/test2.jar|coprocessor.AnotherTest|USER*

path-to-jar可以是一个完整的HDFS地址或其他Hadoop HleSystem类支持的地址或者本地地址

classrmme定义了具体的实现类

priority 只能是 SYSTEM 或 USER

不要在协处理器定义中添加空格。解释十分严格，会使配置无效

Configuration conf = HBaseConfiguration.create();

FileSystem fs = FileSystem.get(conf);
Path path = new Path(fs.getUri()+Path.SEPARATOR+"test.jar");//要一个存在的协处理器

HTableDescriptor htd = new HTableDescriptor("testtable1");
htd.addFamily(new HColumnDescriptor("coifaml"));
htd.setValue("COPROCESSOR$1",path.toString() +
"|" + RegionObserver.class.getCanonicalName() +
"|" + Coprocessor.PRIORITY_USER);

HBaseAdmin admin = new HBaseAdmin(conf);
admin.createTable(htd);
System.out.println(admin.getTableDescriptor(Bytes.toBytes("testtable1")));

表被启用，且 region被打开，框架会首先加载配置文件中定义的协处理器，然后再加载表描述符中的协处理器

RegionObserver 类

region级别的Coprocessor子类RegionObserver类， 它属于observer协处理器：当一个特定的region级别的操作发生时，它们的钩子 函数会被触发

操作分为两类：region生命周期变化和客户端API调用

处理region生命周期事件



这些observer可以与pending open、open和 pending close状态通过钩子链接。每一个钩子都被框架隐式地调用

observer调用时，所有的调用都有一个特定的第一参数：
ObserverContext<RegionCoprocessorEnvironment> c
特殊的CoprocessorEnvironment包装让用户可以控制在钩子执行之后会发生什么

参阅 “RegionCoprocessorEnvironment 类”和 “ObserverContext 类” 两节中的详细介绍
所有参数和异常都被忽略了。可以从线上文档中得到完整的定义。需要注意的是，

状态：pending open。region将要被打开时会处于这个状态。监听的协处理器可以搭载这个过程或阻止这个过程

void preOpen(...)/ void postOpen(...)

这些方法会在region被打开前或刚刚打开后被调用。用户可以在自己的协处理器实现 中使用这两个方法，例如，使用PreOpen()方法告知框架这次打开操作应当被放弃，或勾住postOpen ()方法来触发一次缓存预热或其他一些操作

region经过pending open,且在打开状态之前，region服务器可能需要从WAL中应用 一些记录到region中，这时会触发以下方法：

void preWALRestore(...)/ void postWALRestore(...)

这个方法可以细粒度地控制在WAL重做进行哪些操作。用户可以访问修改记录，因此用户就可以监督哪些记录被实施了。

状态：open。当一个region被部署到一个region服务器中，正常工作时，这个 region会被认为处于open状态。此前本书中提到的那些方法就可以被应用在这个region 上了，例如，region的内存存储可以被持久化到磁盘，当它变得非常大时，region也可以被拆分。可用的钩子函数如下：

void preFlush (...) / void postFlush(… ）
void preCompact(...) / void postCompact(...)
void preSplit(...) / void postSplit(...)

pre方法在事件执行前被调用，post方法在事件执行后被调用。例如，用户使用PreSplit ()钩子函数可以有效地禁用region拆分，然后手动执行这些操作。

状态：pending close。最后 一 组 region监听器的钩子函数可以监听pending close状态。 这个状态在region状态从open到 closed转变时发生。在 region被关闭之前和之后，以 下钩子函数将被执行：

void preClose(..., boolean abortRequested)/ void postClose(..., boolean abortRequested)

abortRequested参数包含了 region被关闭的原因。通常情况下，region会在正常 操作中被关闭，例如，region由于负载均衡被移动到其他region服务器时被关闭。 也有可能是由于region服务器被撤销，且需避免一些副作用。当这些情况发生时， 所有它管理的region都会被撤销，同时用户从这个参数中可以看到是否符合这种情况。

处理客户端API事件

所有客户端API调用都显式地从客户端应用中传输到 region服务器。用户可以在这些调用执行前或刚刚执行后拦截它们

HTable操作前后相关
void preGet(...)/ void postGet(...)
在客户端HTable. get ()请求执行之前和之后调用
boolean preExists(...)/ boolean postExists(...)
void preincrement(...)/ void postincrement (...)
void preDelete(...)/ void postDelete(...)
......

RegionCoprocessorEnvironment类

实现RegionObserver类的协处理器环境的实例是基于RegionCoprocessor Environment类的， RegionCoprocessorEnvironment 实现了 Coprocessor Environment 接口



getRegion()方法可得到目前正在管理的HRegion实例，同时可以执行这个类提 供的方法

用户代码可访问共享的RegionServerServices实例



ObserverContext类

RegionObserver类提供的所有回调函数都需要一个特殊的上下文作为共同的参数： ObserverContext类

它提供了访问当前系统环境的入口，同时添加了一些关键功能用以通知协处理器框架在回调函数完成时需要做什么

所有的协处理器在执行时共用一个上下文实例，并会随着环境一起变化



两个重要的上下文方法是bypass()和 completeO。为协处理器实现提供了选择，以控制框架后续行为

complete ()的调用会影响后面执行的协处理器

bypass ()方法可以停止当前服务进程的处理过程

//使用bypass停止region的自动拆分：
Override
public void preSplit(ObserverContext<RegionCoprocessorEnvironment> e){
e.bypass();
}

与基于接口实现自己的RegionObserver相比，用户可以使用基类修改自己需要的部分

BaseRegionObserve类

这个类作为实现监听类型协处理器的基类。它实现了所有RegionObserver 接口的空方法。用户需要重载他们来实现自己的功能

//服务器端的协处理器代码
//检查特殊get请求的region observer
public class RegionObserverExample extends BaseRegionObserver{
public static final byte[] FIXED_ROW = Bytes.toBytes("@@@GETTIME@@@");

@Override
public void preGetOp(final ObserverContext<RegionCoprocessorEnvironment> e,
final Get get, final List<Cell> results) throws IOException {

if(Bytes.equals(get.getRow(),FIXED_ROW)) {
KeyValue kv = new KeyValue(get.getRow(),FIXED_ROW,FIXED_ROW, Bytes.toBytes(System.currentTimeMillis()));
results.add(kv);
}

}

}
//加载到配置文件,value值前面的是包名，后面的是类名
<property>
<name>hbase.coprocessor.region.classes</name>
<value>customF.RegionObserverExample</value>
</property>
//编译后的JAR添加到了 hbase-env.sh的 HBASE_ CLASSPATH中，region服务器在JRE中可以加载这个类
export HBASE_CLASSPATH="/home/zys/hbase/lib/RegionServerE.jar"

这些get请求都针对已经存在的表，因为不存在的表会产生错误

创建新表后，向表添加一行数据，这一行数据随后也被检索出来了。可以观察到添加的列与表中实际的数据混在了结果中。为了避免这种情况需要使用bypass

results.add(kv);
e.bypass () ;
//一旦特殊的kv被添加时之后的操作就跳过，终止当前的进程的处理过程

//1.x版本的方法可能更改了名字，如下
interface RegionObserver extends Coprocessor
class BaseRegionObserve implement RegionObserver
RegionObserverExample extends BaseRegionObserver{
//方法
@Override
public void preGetOp(final ObserverContext<RegionCoprocessorEnvironment> e,
final Get get, final List<Cell> results)

//ObserverContext保证协处理器生命周期内的状态
//RegionCoprocessorEnvironraent提供执行时的环境参数，系统入口
}

MasterObserver 类

Coprocessor的第二个子类是为了处理master服务器的所有回调函数。（这些操作和API调用会在管理功能介绍），与关系型数据库中DDL（数据库模式定义语言）类似，它们可以被归类到数据处理操作中。基于上述原因MasterObserver类提供如下钩子函数

void preCreateTable(...)/ void postCreateTable(...)
在表创建前后被调用。
void preDeleteTable(...)/ void postDeleteTable(...)
在表删除前后被调用。
void preModifyTable(...)/ void postModifyTable(...)
在表修改前后被调用。
void preAddColumn(...)/ void postAddColumn(...)
在表中添加列前后被调用。
void preModifyColumn(...)/ void postModifyColumn(...)
在表中列被修改前后被调用。
void preDeleteColumn(...)/ void postDeleteColumn(...)
在表中列被删除前后被调用。
void preEnableTable (...)/ void postEnableTable (… ）
在表启用前后被调用。
void preDisableTable(...)/ void postDisableTable(...)
在表禁用前后被调用。
void preMove(...)/ void postMove(...)
在region被移动前后被调用。
void preAssign(...)/ void postAssign(...)
在region分配前后被调用。
void preUnassign(...)/ void postUnassign(...)
在region未分配前后被调用。
void preBalance(...)/ void postBalance(...)
在region负载均衡操作前后被调用。
boolean preBalanceSwitch(...)/ void postBalanceSwitch(...)
在修改自动负载均衡标志位前后被调用。
void preShutdown(...)
在集群关闭工作开始前被调用。没有post钩子函数，因为集群关闭之后没有进程
客户端A P I:高级特性 181
可以执行post函数。
void preStopMaster(...)
在 master进程停止工作开始前被调用。没有post钩子函数，因为master停止工作 之后没有进程可以执行post函数

MasterCoprocessorEnvironment 类

MasterCoprocessorEnvironment 封装了 MasterObserver 实例，它同样实现了 CoprocessorEnvironment接口，因此它也能提供getTable ()之类的方法帮助用户在自己 的实现中访问数据

//MasterCoprocessorEnvironment 类提供的非继承方法
MasterServices getMasterServices() 提供可访问的共享MasterServices实例

BaseMasterObserver类

可直接实现MasterObserver接口，或扩展BaseMasterObserver类来实现自己的

功能

public class MasterObserverExample extends BaseMasterObserver {
@0verride
public void postCreateTable(ObserverContext<MasterCoprocessorEnvironment> env, HRegionlnfo[] regions,boolean sync)throws IOException {
//从表描述符中得到表名
String tableName = regions[0].getTableDesc().getNameAsString();
MasterServices services=env.getEnvironment().getMasterServices();
//获取可用的服务，同时取得真实文件系统的引用
MasterFileSystem masterFileSystem=services.getMasterFileSystem();
FileSystem fileSystem = masterFileSystem.getFileSystem();
//创建新目录用来存储客户端应用的二进制数据
Path blobPath = new Path(tableName + "-blobs");
fileSystera.mkdirs(blobPath);

用户需要将master配置项添加到hbase-site.xml文件中，然后协处理器将被master进程加载：

协处理器一旦启动成功就会立即进行监听事件并自动执行添加的代码

因为Environment实例被ObserverContext封装过，用户也可以调用流程控制函数 bypass ()和 complete ()。用户可以使用它们显式地禁用一些操作，或跳过后续要执行的协处理器