如果要统对hbase中的数据，进行某种统计，比如统计某个字段最大值，统计满足某种条件的记录数，统计各种记录特点，并按照记录特点分类（类似于sql的group by）~
常规的做法就是把hbase中整个表的数据scan出来，或者稍微环保一点，加一个filter，进行一些初步的过滤（对于rowcounter来说，就加了FirstKeyOnlyFilter），但是这么做来说还是会有很大的副作用，比如占用大量的网络带宽（当标级别到达千万级别，亿级别之后）尤为明显，RPC的量也是不容小觑的。
拿row counter这个简单例子来说，我要统计总行数，如果每个region 告诉我他又多少行，然后把结果告诉我，我再将他们的结果汇总一下，不就行了么？
现在的问题是hbase没有提供这种接口，来统计每个region的行数，那是否我们可以自己来实现一个呢？
没错，正如本文标题所说，我们可以自己来实现一个Endpoint，然后让hbase加载起来，然后我们远程调用即可。
先弄清楚什么是hbase coprocessor
hbase有两种coprocessor，一种是Observer（观察者），类似于关系数据库的trigger（触发器），另外一种就是EndPoint，类似于关系数据库的存储过程。
观察者这里就多做介绍了，这里介绍Endpoint。
EndPoint是动态RPC插件的接口，它的实现代码被部署在服务器端（regionServer），从而能够通过HBase RPC调用。客户端类库提供了非常方便的方法来调用这些动态接口，它们可以在任意时候调用一个EndPoint，它们的实现代码会被目标region远程执行，结果会返回到终端。用户可以结合使用这些强大的插件接口，为HBase添加全新的特性。
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调用。
如图 
|
1
2
3
|
public interface CounterProtocol extends CoprocessorProtocol {
public long count( byte [] start, byte [] end) throws IOException;
} |
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
public class CounterEndPoint extends BaseEndpointCoprocessor implements CounterProtocol {
@Override
public long count( byte [] start, byte []end) throws IOException {
// aggregate at each region
Scan scan = new Scan();
long numRow = 0 ;
InternalScanner scanner = ((RegionCoprocessorEnvironment) getEnvironment()).getRegion()
.getScanner(scan);
try {
List<KeyValue> curVals = new ArrayList<KeyValue>();
boolean hasMore = false ;
do {
curVals.clear();
hasMore = scanner.next(curVals);
if (Bytes.compareTo(curVals.get( 0 ).getRow(), start)< 0 ) {
continue ;
}
if (Bytes.compareTo(curVals.get( 0 ).getRow(), end)>= 0 ) {
break ;
}
numRow++;
} while (hasMore);
} finally {
scanner.close();
}
return numRow;
}
} |
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
public class CounterEndPointDemo {
public static void main(String[] args) throws IOException, Throwable {
final String startRow = args[ 0 ];
final String endRow = args[ 1 ];
@SuppressWarnings ( "resource" )
HTableInterface table = new HTable(HBaseConfiguration.create(), "tc" );
Map< byte [], Long> results;
// scan: for all regions
results = table.coprocessorExec(CounterProtocol. class , startRow.getBytes(),
endRow.getBytes(), new Batch.Call<CounterProtocol, Long>() {
public Long call(CounterProtocol instance) throws IOException {
return instance.count(startRow.getBytes(), endRow.getBytes());
}
});
long total = 0 ;
for (Map.Entry< byte [], Long> e : results.entrySet()) {
System.out.println(e.getValue());
total += e.getValue();
}
System.out.println( "total:" + total);
}
} |
整个程序的框架其实又是另外一个mapreduce，只是运行在region server上面，reduce运行在客户端，其中map计算量较大，reduce计算量很小！
另外需要提醒的是：
protocol的返回类型，可以是基本类型。
如果是一个自定义的类型需要实现org.apache.hadoop.io.Writable接口。
关于详细的支持类型，请参考代码hbase源码：org.apache.hadoop.hbase.io.HbaseObjectWritable
1. 通过hbase-site.xml增加
|
1
2
3
4
|
< property >
< name >hbase.coprocessor.region.classes</ name >
< value >xxxx.CounterEndPoint </ value >
</ property >
|
2. 通过shell方式
增加：
|
1
2
3
4
5
6
|
hbase(main):005:0> alter 't1' , METHOD => 'table_att' ,
Updating all regions with the new schema...1 /1 regions updated.
Done.0 row(s) in 1.0730 seconds
|
coprocessor格式为：
[FilePath]|ClassName|Priority|arguments
arguments: k=v[,k=v]+
卸载：
这是一个最简单的例子，另外还有很多统计场景，可以用在这种方式实现，有如下好处：
其他应用场景？
参考：
1. http://blogs.apache.org/hbase/entry/coprocessor_introduction
2. https://issues.apache.org/jira/browse/HBASE-1512