HBASE COPROCESSOR EndPoint实例

问题导读：
1、统计总行数，理想的方式应该是怎样？
2、什么是Endpoint，怎样去实现它 ？
3、有哪几种方式去部署 ？




前言：
如果要统对hbase中的数据，进行某种统计，比如统计某个字段最大值，统计满足某种条件的记录数，统计各种记录特点，并按照记录特点分类（类似于sql的group by）~

常规的做法就是把hbase中整个表的数据scan出来，或者稍微环保一点，加一个filter，进行一些初步的过滤（对于rowcounter来说，就加了FirstKeyOnlyFilter），但是这么做来说还是会有很大的副作用，比如占用大量的网络带宽（当标级别到达千万级别，亿级别之后）尤为明显，RPC的量也是不容小觑的。

理想的方式应该是怎样？
拿row counter这个简单例子来说，我要统计总行数，如果每个region 告诉我他又多少行，然后把结果告诉我，我再将他们的结果汇总一下，不就行了么？
现在的问题是hbase没有提供这种接口，来统计每个region的行数，那是否我们可以自己来实现一个呢？
没错，正如本文标题所说，我们可以自己来实现一个Endpoint，然后让hbase加载起来，然后我们远程调用即可。

什么是Endpoint？
先弄清楚什么是hbase coprocessor

hbase有两种coprocessor，一种是Observer（观察者），类似于关系数据库的trigger（触发器），另外一种就是EndPoint，类似于关系数据库的存储过程。

观察者这里就多做介绍了，这里介绍Endpoint。

EndPoint是动态RPC插件的接口，它的实现代码被部署在服务器端（regionServer），从而能够通过HBase RPC调用。客户端类库提供了非常方便的方法来调用这些动态接口，它们可以在任意时候调用一个EndPoint，它们的实现代码会被目标region远程执行，结果会返回到终端。用户可以结合使用这些强大的插件接口，为HBase添加全新的特性。

怎么实现一个EndPoint
1. 定义一个新的protocol接口，必须继承CoprocessorProtocol.
2. 实现终端接口，继承抽象类BaseEndpointCoprocessor，改实现代码需要部署到
3. 在客户端，终端可以被两个新的HBase Client API调用 。单个region：HTableInterface.coprocessorProxy(Class<T> protocol, byte[] row) 。rigons区域：HTableInterface.coprocessorExec(Class<T> protocol, byte[] startKey, byte[] endKey, Batch.Call<T,R> callable)，这里的region是通过一个row来标示的，就是说，改row落到那个region，RPC就发给哪个region，对于start-end的，[start,end)范围内的region都会受到RPC调用。


如下图所示：
<ignore_js_op style="word-wrap: break-word; color: rgb(68, 68, 68); font-family: Tahoma, 'Microsoft Yahei', Simsun; font-size: 14.3999996185303px; line-height: 16.7999992370605px;"> 

1. public interface CounterProtocol extends CoprocessorProtocol {
2.         public long count(byte[] start, byte[] end) throws IOException;
3. }

复制代码

1. public class CounterEndPoint extends BaseEndpointCoprocessor implements CounterProtocol {
3.         @Override
4.         public long count(byte[] start, byte []end) throws IOException {
5.                 // aggregate at each region
6.                 Scan scan = new Scan();
7.                 long numRow = 0;
9.                 InternalScanner scanner = ((RegionCoprocessorEnvironment) getEnvironment()).getRegion()
10.                                 .getScanner(scan);
11.                 try {
12.                         List<KeyValue> curVals = new ArrayList<KeyValue>();
13.                         boolean hasMore = false;
14.                         do {
15.                                 curVals.clear();
16.                                 hasMore = scanner.next(curVals);
17.                                 if (Bytes.compareTo(curVals.get(0).getRow(), start)<0) {
18.                                         continue;
19.                                 }
20.                                 if (Bytes.compareTo(curVals.get(0).getRow(), end)>= 0) {
21.                                         break;
22.                                 }
23.                                 numRow++;
24.                         } while (hasMore);
25.                 } finally {
26.                         scanner.close();
27.                 }
28.                 return numRow;
29.         }
31. }

复制代码

1. public class CounterEndPointDemo {
2.         public static void main(String[] args) throws IOException, Throwable {
3.                 final String startRow = args[0];
4.                 final String endRow = args[1];
6.                 @SuppressWarnings("resource")
7.                 HTableInterface table = new HTable(HBaseConfiguration.create(), "tc");
8.                 Map<byte[], Long> results;
10.                 // scan: for all regions
11.                 results = table.coprocessorExec(CounterProtocol.class, startRow.getBytes(),
12.                                 endRow.getBytes(), new Batch.Call<CounterProtocol, Long>() {
13.                                         public Long call(CounterProtocol instance) throws IOException {
14.                                                 return instance.count(startRow.getBytes(), endRow.getBytes());
15.                                         }
16.                                 });
18.                 long total = 0;
19.                 for (Map.Entry<byte[], Long> e : results.entrySet()) {
20.                         System.out.println(e.getValue());
21.                         total += e.getValue();
22.                 }
24.                 System.out.println("total:" + total);
25.         }
26. }

复制代码

整个程序的框架其实又是另外一个mapreduce，只是运行在region server上面，reduce运行在客户端，其中map计算量较大，reduce计算量很小！

另外需要提醒的是：
protocol的返回类型，可以是基本类型。
如果是一个自定义的类型需要实现org.apache.hadoop.io.Writable接口。
关于详细的支持类型，请参考代码hbase源码：org.apache.hadoop.hbase.io.HbaseObjectWritable

怎么部署？
1. 通过hbase-site.xml增加

1. <property>
2.   <name>hbase.coprocessor.region.classes</name>
3.   <value>xxxx.CounterEndPoint </value>
4. </property>

复制代码

1、如果要配置多个，就用逗号（,)分割。
2、包含此类的jar必须位于hbase的classpath
3、这种coprocessor是作用于所有的表，如果你只想作用于部分表，请使用下面一种方式。

2、 通过shell方式
增加：

1. hbase(main):005:0> alter 't1', METHOD => 'table_att',
2. 'coprocessor'=>'hdfs:///foo.jar|com.foo.FooRegionObserver|1001|arg1=1,arg2=2'
3. Updating all regions with the new schema...
4. 1/1 regions updated.
5. Done.
6. 0 row(s) in 1.0730 seconds

复制代码

coprocessor格式为：
[FilePath]|ClassName|Priority|arguments
arguments: k=v[,k=v]+

其中FilePath是hdfs路径，例如/tmp/zhenhe/cp/zhenhe-1.0.jar
ClassNameEndPoint实现类的全名
Priority为，整数，框架会根据这个数据决定多个cp的执行顺序
Arguments，传给cp的参数
如果hbase的classpath包含改类，FilePath可以留空


卸载：
先describe “tableName‘，查看你要卸载的cp的编号
然后alter 't1', METHOD => 'table_att_unset', NAME=> 'coprocessor$3'，coprocessor$3可变。
应用场景
这是一个最简单的例子，另外还有很多统计场景，可以用在这种方式实现，有如下好处：


节省网络带宽
减少RPC调用（scan的调用随着CacheSzie的变小而线性增加），减轻hbase压力
可以提高统计效率，那我之前写过的一个groupby类型的例子来说，大约可以提高50%以上的统计速度。
其他应用场景？

一个保存着用户信息的表，可以统计每个用户信息（counter job）
统计最大值，最小值，平均值，参考：官网
批量删除记录，批量删除某个时间戳的记录