HBase 1.x Coprocessor使用指南
@(HBASE)[hbase]
HBase在0.92版本之后，提供了协处理器功能。
在之前介绍过，HBase提供了过滤器，以减少从服务器返回客户端的数据。而协处理器用于将部分处理工作交由RegionServer处理，而不是全部返回client再处理。
举个例子，HBase的安全机制就是通过协处理器实现的。当用户向HBase发出一个读写请求时，HBase会首先触发这个协处理器，它会在读写操作前确认用户是否有这个权限。
一、概述
（以下概述性的内容摘自网络）。
1、起因(Why HBase Coprocessor)
HBase作为列族数据库最经常被人诟病的特性包括：无法轻易建立“二级索引”，难以执行求和、计数、排序等操作。比如，在旧版本的(<0.92)Hbase中，统计数据表的总行数，需要使用Counter方法，执行一次MapReduce Job才能得到。虽然HBase在数据存储层中集成了MapReduce，能够有效用于数据表的分布式计算。然而在很多情况下，做一些简单的相加或者聚合计算的时候，如果直接将计算过程放置在server端，能够减少通讯开销，从而获得很好的性能提升。于是，HBase在0.92之后引入了协处理器(coprocessors)，实现一些激动人心的新特性：能够轻易建立二次索引、复杂过滤器(谓词下推)以及访问控制等。
2、灵感来源( Source of Inspration)
HBase协处理器的灵感来自于Jeff Dean 09年的演讲( P66-67)。它根据该演讲实现了类似于bigtable的协处理器，包括以下特性:
- 每个表服务器的任意子表都可以运行代码
- 客户端的高层调用接口(客户端能够直接访问数据表的行地址，多行读写会自动分片成多个并行的RPC调用)
- 提供一个非常灵活的、可用于建立分布式服务的数据模型
- 能够自动化扩展、负载均衡、应用请求路由
HBase的协处理器灵感来自bigtable，但是实现细节不尽相同。HBase建立了一个框架，它为用户提供类库和运行时环境，使得他们的代码能够在HBase region server和master上处理。
3、细节剖析（Implementation)
协处理器分两种类型，系统协处理器可以全局导入region server上的所有数据表，表协处理器即是用户可以指定一张表使用协处理器。协处理器框架为了更好支持其行为的灵活性，提供了两个不同方面的插件。一个是观察者（observer），类似于关系数据库的触发器。另一个是终端(endpoint)，动态的终端有点像存储过程。
3.1观察者(Observer)
观察者的设计意图是允许用户通过插入代码来重载协处理器框架的upcall方法，而具体的事件触发的callback方法由HBase的核心代码来执行。协处理器框架处理所有的callback调用细节，协处理器自身只需要插入添加或者改变的功能。
HBase 提供了三种观察者接口：
- RegionObserver：提供客户端的数据操纵事件钩子：Get、Put、Delete、Scan等。
- WALObserver：提供WAL相关操作钩子。
- MasterObserver：提供DDL-类型的操作钩子。如创建、删除、修改数据表等。
这些接口可以同时使用在同一个地方，按照不同优先级顺序执行.用户可以任意基于协处理器实现复杂的HBase功能层。HBase有很多种事件可以触发观察者方法，这些事件与方法从HBase0.92版本起，都会集成在HBase API中。不过这些API可能会由于各种原因有所改动，不同版本的接口改动比较大，具体参考Java Doc。
3.2终端(Endpoint)
终端是动态RPC插件的接口，它的实现代码被安装在服务器端，从而能够通过HBase RPC唤醒。客户端类库提供了非常方便的方法来调用这些动态接口，它们可以在任意时候调用一个终端，它们的实现代码会被目标region远程执行，结果会返回到终端。用户可以结合使用这些强大的插件接口，为HBase添加全新的特性。
具体使用方法参考下面。
二、Observer
（一）示例
1、准备类文件
这个例子的coprocessor是一个RegionObserver，它判断如果请求的rowkey是@@@GETTIME@@@，则返回系统当前时间，然后不再请求region读取实际的数据(e.bypass())，否则，有可能返回2行。
幅使用的是preGetOp()方法，因此所有的Get操作都会先经过这个Coprocessor处理。
public class CoprocessorDemo extends BaseRegionObserver{
public static final byte[] FIXED_ROW = Bytes.toBytes("@@@GETTIME@@@");
@Override
public void preGetOp(ObserverContext<RegionCoprocessorEnvironment> e, Get get, 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);
e.bypass();
}
}
}
2、部署方式一
这种方法适用于集群管理人员使用，所部署的coprocessor会影响所有表，所有region。
（1）修改配置，增加协处理器类
<property>
<name>hbase.coprocessor.region.classes</name>
<value>org.apache.hadoop.hbase.security.token.TokenProvider,org.apache.hadoop.hbase.security.access.AccessController,com.lujinhong.commons.hbase.coprocessor.CoprocessorDemo</value>
</property>
这里也可以看出来，安全相关的控制都使用协处理器完成的。
（2）将包含上述类的jar包入到hbase的lib目录中
（3）重启hbase
bin/rolling-restart.sh
3、部署方式二
这种方式适合开发人员使用，只会影响特写的表或者region。这种方式有可能导致开发人员滥用coprocessor，从而使得hbase集群负载过高，因此建议回收建表权限，只能由集群管理人员建表，并在建表时指定coprocessor。
这种方式无须重启集群，从而达到热加载的目的。
（1）将包含上述类的jar包放到某个hdfs路径
如/hbase/userlib。当然也可以直接放在本地目录，但要保证每台hbase服务器都有这个类。
（2）创建表时，指定coprocessor
create 'ljhtest', 'f1'
disable 'ljhtest'
alter 'ljhtest', 'Coprocessor'=>'hdfs://testing/hbase/userlib/gdc-commons-hbase-0.1-SNAPSHOT.jar|com.lujinhong.commons.hbase.coprocessor.CoprocessorDemo|1073741825|arg1=1'
enable 'ljhtest'
注意jar包的权限，如果hbase用户不能读取这个jar包，会导致enable时失败。
说明文档如下：
hbase> alter 't1',
'coprocessor'=>'hdfs:///foo.jar|com.foo.FooRegionObserver|1001|arg1=1,arg2=2'
Since you can have multiple coprocessors configured for a table, a
sequence number will be automatically appended to the attribute name
to uniquely identify it.
The coprocessor attribute must match the pattern below in order for
the framework to understand how to load the coprocessor classes:
[coprocessor jar file location] | class name | [priority] | [arguments]
也可以在java代码中通过HTableDescription来指定coprocessor。
4、使用coproccessor
（1）在hbase shell中使用coprocessor
hbase(main):007:0* get 'ljhtest3','@@@GETTIME@@@'
COLUMN CELL
@@@GETTIME@@@:@@@GETTIME@@@ timestamp=9223372036854775807, value=\x00\x00\x01T\xCE0=\x9E
1 row(s) in 0.1610 seconds
就会返回当前时间。注意如果使用方式一部署的话，请求所有表均会返回正确结果，而使用方式二部署的话只有请求指定的表才会返回当前时间。
（2）使用java API使用coprocessor
一样的，没有什么特写，正常读取即可。
5、问题
在运行coprocessor时，若不失效，则到region所在的regionserver中查看日志，比如有些包没打包上去等。
三、Endpoint
完整代码请见：https://github.com/lujinhong/lujinhong-commons/tree/master/lujinhong-commons-hbase/src/main/java/com/lujinhong/commons/hbase/coprocessor
除了本例以外，还可以参考hbase源代码中的RowCountEndpoint。
注意，hbase 0.98对实现endpoint的API作了很大的调整，《hbase权威指南》等书的API均不能再使用。
创建一个Endpoint的基本流程可以归纳为：
（1）创建一个通信协议：准备一个proto文件，然后使用protoc工具来生成协议类文件。这个文件需要在服务端及客户端存在。
（2）创建一个Service类，实现具体的业务逻辑
（3）创建表时指定使用这个EndPoint，或者是全局配置。
（4）创建一个Client类，调用这个RPC方法。
（零）业务常景描述
HBase表中有一个family, 2相column，分别为f:c1, f:c2。rowkey为某个用户id（当然经过hash以后以避免热点），2个列分别表示这个用户在2款产品的在线时间，单位为秒。如：
id1 column=f1:c1, timestamp=1464323601847, value=500000
id1 column=f1:c2, timestamp=1464323601883, value=600000
id2 column=f1:c1, timestamp=1464323648768, value=500
id2 column=f1:c2, timestamp=1464323648758, value=600000 id3 column=f1:c1, timestamp=1464323648775, value=700000
id3 column=f1:c2, timestamp=1464323648783, value=700
id4 column=f1:c1, timestamp=1464324774802, value=700000
id4 column=f1:c2, timestamp=1464324774845, value=800000
下面就基于这些数据来计算。
要求计算：
（1）2个产品的真实用户有多少，定义为在线时长超过10分钟的
（2）这2个产品的真实用户平均在线时长是多少
协处理器的处理逻辑为：
（1）请求为这2个列名，万一上线新产品时可以直接使用，也可示范如何定义一个request。
（2）计算2个产品的真实用户count1, count2
（3）计算这2个产品用户分别的在线总时长sum1, sum2
（4）将各个region的sum和count分别加起来后，计算2个产品的平均值。
（一）准备proto文件
请求参数为列的名称，用“；”分隔，返回的是这2个产品的真实用户数与在线总时长。
option java_package = "com.lujinhong.coprocessor";
option java_outer_classname = "MultiColumnSumProtocol";
option java_generic_services = true;
option java_generate_equals_and_hash = true;
option optimize_for = SPEED;
message CountRequest {
required string columns = 1;
}
message CountResponse {
required int64 count1 = 1 [default = 0];
required int64 count2 = 2 [default = 0];
required int64 sum1 = 3 [default = 0];
required int64 sum2 = 4 [default = 0];
}
service RowCountService {
rpc getCountAndSum(CountRequest)
returns (CountResponse);
}
（二）使用protoc生成类文件
protoc --java_out=../java/ MultiColumnSum.proto
这个命令在使用上面的proto文件生成相应的类文件，这个类文件有几个地方需要注意：
1、生成了一个CountRequest内部类，表示请求信息
2、生成了一个CountResponse内部类，表示返回信息
3、生成了一个 RowCountService内部类，表示所提供的服务，这个类还有一个内部接口，这个接口定义了 getCountAndSum()这个方法。
我们下面需要做的就是实现这个接口的这个方法，提供真正的服务。
（三）实现真实的服务
1、类的结构
提供真实服务的类继承自上面自动生成的Server类，同时需要实现Coprocessor和CoprocessorService2个接口：
public class MultiColumnSum extends MultiColumnSumProtocol.RowCountService implements Coprocessor, CoprocessorService
它需要实现以下4个方法，下面我们逐一讨论一下：
@Override
public Service getService() {
return null;
}
@Override
public void start(CoprocessorEnvironment env) throws IOException {
}
@Override
public void stop(CoprocessorEnvironment env) throws IOException {
}
@Override
public void getCountAndSum(RpcController controller, CountRequest request, RpcCallback<CountResponse> done) {
}
2、 getService()
这个方法直接返回自身即可。
@Override
public Service getService() {
return this;
}
3、 start(CoprocessorEnvironment env)
这个方法会在coprocessor启动时调用，这里判断了是否在一个region内被使用，而不是master，WAL等环境下被调用。
@Override
public void start(CoprocessorEnvironment env) throws IOException {
if (env instanceof RegionCoprocessorEnvironment) {
this.env = (RegionCoprocessorEnvironment) env;
} else {
throw new CoprocessorException("Must be loaded on a table region!");
}
}
4、stop(CoprocessorEnvironment env)
这个方法会在coprocessor完成时被调用，可用于关闭资源等，这里为空。
@Override
public void stop(CoprocessorEnvironment env) throws IOException {
}
5、 getCountAndSum(…)
这是整个类的核心方法，用于实现真正的业务逻辑。关键的步骤有：
（1）根据request创建一个Scanner，然后使用它创建一个 InternalScanner，可以更高效的进行scan
（2）对扫描出来的行进行分析处理，将结果保存在几个变量中。
（3）调用response的各个set()方法，设置返回的结果。
（4）使用 done.run(response); 返回结果到客户端。
这个方法的完整代码如下：
@Override
public void getCountAndSum(RpcController controller, CountRequest request, RpcCallback<CountResponse> done) {
long[] values = { 0, 0, 0, 0 };
String columns = request.getColumns();
if (columns == null || "".equals(columns))
throw new NullPointerException("you need specify the columns");
String[] columnArray = columns.split(";");
Scan scan = new Scan();
for (String column : columnArray) {
scan.addColumn(Bytes.toBytes(FAMILY), Bytes.toBytes(column));
}
MultiColumnSumProtocol.CountResponse response = null;
InternalScanner scanner = null;
try {
scanner = env.getRegion().getScanner(scan);
List<Cell> results = new ArrayList<Cell>();
boolean hasMore = false;
do {
hasMore = scanner.next(results);
if (results.size() < 2)
continue;
Cell kv0 = results.get(0);
long value1 = Long.parseLong(Bytes.toString(CellUtil.cloneValue(kv0)));
Cell kv1 = results.get(1);
long value2 = Long.parseLong(Bytes.toString(CellUtil.cloneValue(kv1)));
if(value1 > 60000){
values[0] += 1;
values[2] += value1;
}
if(value2 > 60000){
values[1] += 1;
values[3] += value2;
}
results.clear();
} while (hasMore);
// 生成response
response = MultiColumnSumProtocol.CountResponse.newBuilder().setCount1(values[0]).setCount2(values[1]).setSum1(values[2]).setSum2(values[3]).build();
} catch (IOException e) {
e.printStackTrace();
ResponseConverter.setControllerException(controller, e);
} finally {
if (scanner != null) {
try {
scanner.close();
} catch (IOException ignored) {
}
}
}
done.run(response);
}
（四）部署coprocessor
将上述2个类进行打包，然后按照上面Oberver部分介绍的部署方法来部署coprocessor。
（五）客户端使用coprocessor
注意，如果很多代码用到这个coprocessor，最好封装成更方便调用的方式。
最核心的代码是：
Map<byte[], ResponseInfo> map = table.coprocessorService(MultiColumnSumProtocol.RowCountService.class, null,
null, new Batch.Call<MultiColumnSumProtocol.RowCountService, ResponseInfo>() {
@Override
public ResponseInfo call(MultiColumnSumProtocol.RowCountService service) throws IOException {
BlockingRpcCallback<MultiColumnSumProtocol.CountResponse> rpcCallback = new BlockingRpcCallback<>();
service.getCountAndSum(null, request, rpcCallback);
MultiColumnSumProtocol.CountResponse response = rpcCallback.get();
ResponseInfo responseInfo = new ResponseInfo();
responseInfo.count1 = response.getCount1();
responseInfo.count2 = response.getCount2();
responseInfo.sum1 = response.getSum1();
responseInfo.sum2 = response.getSum2();
return responseInfo;
}
});
将调用的结果返回保存在一个map中，每个region会产生一条数据。然后通过合并各个region的结果来得出最终的结果即可。
ResponseInfo result = new ResponseInfo();
for (ResponseInfo ri : map.values()) {
result.count1 += ri.count1;
result.count2 += ri.count2;
result.sum1 += ri.sum1;
result.sum2 += ri.sum2;
}
System.out.println("Produce 1 has " + result.count1 + " user, all online time is " + result.sum1 / 1000
+ " minutes, average online time is " + result.sum1 / 1000 / result.count1 + "minutes.");
System.out.println("Produce 2 has " + result.count2 + " user, all online time is " + result.sum2 / 1000
+ " minutes, average online time is " + result.sum2 / 1000 / result.count2 + "minutes.");
（六）运行程序
打包并上传至集群，然后运行程序。
注意只打包client类与protocol类，不要打包Service类。即部署到集群的jar包包括Service类和protocol类，而运行任务的jar包包括client类与protocol类。
hadoop jar lujinhong-commons-hbase-0.1-SNAPSHOT.jar com.lujinhong.commons.hbase.coprocessor.MultiColumnSumClient
输出结果为：
Produce 1 has 3 user, all online time is 1900 minutes, average online time is 633minutes.
Produce 2 has 3 user, all online time is 2000 minutes, average online time is 666minutes.