这几天为了满足业务需求，自己写了个Coprocessor，这里写一篇博客记录一下。

使用Coprocessor的目的是这样的，假如你的业务使你不得不进行全表查询，如果使用传统的Scan的话，那么全表查询将会给集群带来高的带宽压力，而且可能Client端也负载不了海量数据的计算。HBase提供了AggregateImplementation，可以进行简单的例如计算sum、average等操作，但是这些操作都是只针对一个列进行的，我遇到的情况是需要同时统计多个列，因此，就只能自己写Coprocessor了。

HBase的Coprocessor是运行在RegionServer上的一段程序，有点类似于关系型数据库中的触发器和存储过程。这个Coprocessor通常有两种类型，分别是Observer和Endpoint。Oberserver相当于触发器，可用于建立二级索引等操作。Endpoint相当于存储过程，可用于在各个RegionServer上做一些计算等，然后将计算的结果汇集到Client端来做最后的处理，这有点儿像Map/Reduce的过程。我这几天是将Coprocessor用作Endpoint，所以这里只对编写Endpoint的代码进行介绍。

在HBase版本0.96以后，HBase的RPC框架采用的就是protobuf作为通讯协议，关于protobuf的介绍网上有很多，这里就不再介绍了。

tar -zxvf protobuf-2.5.0.tar.gz

./configure --prefix=/home/cyber_space/protobuf

make

make install

这个.proto文件我是仿照hbase源代码中的两个Example写的，这两个Example的源代码路径在如下路径中，

/hbase-1.0.1.1_src/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/RowCountEndpoint.java

/hbase-1.0.1.1_src/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/BulkDeleteEndpoint.java

它们的.proto文件的路径在如下的路径中，

~/hbase-1.0.1.1_src/hbase-examples/src/main/protobuf/Examples.proto 

~/hbase-1.0.1.1_src/hbase-examples/src/main/protobuf/BulkDelete.proto

简单解释一下，message是消息关键字，这里我定义了两个消息，用于Coprocessor交流，CountRequest是RPC调用端发出的请求消息，CountResponse是服务端返回的消息，然后又定义了一个Service，这个service里可以定义一个远程调用的方法，getsizes，然后调用的参数就是CountRequest消息，返回是CountResponse消息。[default=0]表示的是这些变量默认为0。每个变量后的=1、=2表示的是变量的优先级。

option java_package = "com.uestc.coprocessor";

option java_outer_classname = "JAggregateProtocol";

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 qiniu_count = 1 [default = 0];

required int64 qiniu2_count = 2 [default = 0];

required int64 jinshan_count = 3 [default = 0];

optional int64 other_cloud = 4 [default=0];

}

service RowCountService {

  rpc getSizes(CountRequest)

    returns (CountResponse);

}

写完这个消息后，使用我们刚才安装的protoc工具将其编译生成我们需要的Java代码，使用的命令如下：

protoc --java_out=/home/cyber_space/ MultiColumnAggregationService.proto

这样就会生成我们在.proto文件中指定的JAVA类，类中包含了必要的Service和getter、setter方法等。

这个实现类需要继承我们用protoc工具生成的Service类，实现Coprocessor、CoprocessorService接口，以便给HBase的RPC框架调用。start方法我是照着Example写的，这个方法会在HBase启动时，或者是Coprocessor启动时被调用。stop方法可以暂时不实现内容。getService方法返回这个类的对象就可了。getSizes方法就是刚才写在.proto文件的service中的RPC方法。下面详细讲一下getSizes方法。

getSizes方法中可通过request的getter方法获取客户端的消息，这里我就使用了request.getColumns() 获取到CountRequest中定义的columns字符串变量。在getSizes方法中实例化了一个Scan，并设置了若干Filter，来过滤掉不需要的列。通过scanner = env.getRegion().getScanner(scan)方法来实例化一个InternalScanner，这个Scanner比较接近表信息在HBase底层的存储结构。它的next方法返回的是一行的所有Cell，每个Cell是有Key-ColumnFamilly-Qualifier-value组成的，可以用CellUtil类来对Cell做一些常用操作。最后Response对象，可以用构建器模式进行实例化，在RPCCallback中放入done即可。

package com.uestc.coprocessor;

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

import org.apache.hadoop.hbase.Coprocessor;

import org.apache.hadoop.hbase.CoprocessorEnvironment;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.coprocessor.CoprocessorException;

import org.apache.hadoop.hbase.coprocessor.CoprocessorService;

import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;

import org.apache.hadoop.hbase.filter.CompareFilter;

import org.apache.hadoop.hbase.filter.Filter;

import org.apache.hadoop.hbase.filter.FilterList;

import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;

import org.apache.hadoop.hbase.protobuf.ResponseConverter;

import org.apache.hadoop.hbase.regionserver.InternalScanner;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.hbase.Cell;

import org.apache.hadoop.hbase.CellUtil;

import com.google.protobuf.RpcCallback;

import com.google.protobuf.RpcController;

import com.google.protobuf.Service;

import com.uestc.coprocessor.JAggregateProtocol.CountRequest;

import com.uestc.coprocessor.JAggregateProtocol.CountResponse;

public class MultiColumnAggregation extends JAggregateProtocol.RowCountService

implements Coprocessor, CoprocessorService {

public static final String COLUMNNAME_QINIU = "Qiniu";

public static final String COLUMNNAME_JINSHAN = "Jinshan";

public static final String COLUMNNAME_QINIU2 = "Qiniu2";

private RegionCoprocessorEnvironment env;

@Override

public void start(CoprocessorEnvironment arg0) throws IOException {

if (arg0 instanceof RegionCoprocessorEnvironment) {

this.env = (RegionCoprocessorEnvironment) arg0;

} else {

throw new CoprocessorException("Must be loaded on a table region!");

}

}

@Override

public void stop(CoprocessorEnvironment arg0) throws IOException {

}

@Override

public Service getService() {

return this;

}

@Override

public void getSizes(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(";");

// 设置filter，只过滤出我们需要的列

Filter filter = null;

ArrayList<Filter> filters = new ArrayList<>();

for (String column : columnArray) {

filter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("location"),

CompareFilter.CompareOp.EQUAL, Bytes.toBytes(column));

filters.add(filter);

}

filter = new FilterList(FilterList.Operator.MUST_PASS_ONE, filters);

Scan scan = new Scan();

scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("location"));

scan.addColumn(Bytes.toBytes("info"), Bytes.toBytes("chunksize"));

scan.setFilter(filter);

JAggregateProtocol.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;

// chunksize 在前，location在后

Cell kv0 = results.get(0);

long chunksize = Long.parseLong(Bytes.toString(CellUtil.cloneValue(kv0)));

Cell kv1 = results.get(1);

String location = Bytes.toString(CellUtil.cloneValue(kv1));

switch (location) {

case COLUMNNAME_QINIU:

values[0] += chunksize;

break;

case COLUMNNAME_QINIU2:

values[1] += chunksize;

break;

case COLUMNNAME_JINSHAN:

values[2] += chunksize;

break;

default:

break;

}

results.clear();

} while (hasMore);

// 生成response

response = JAggregateProtocol.CountResponse.newBuilder().setQiniuCount(values[0]).setQiniu2Count(values[1])

.setJinshanCount(values[2]).build();

} catch (IOException e) {

e.printStackTrace();

ResponseConverter.setControllerException(controller, e);

} finally {

if (scanner != null) {

try {

scanner.close();

} catch (IOException ignored) {

}

}

}

done.run(response);

}

}

客户端代码通过HBase的RPC框架调用HRegionServer上的代码，HRegionServer上的代码完成工作后将结果返回给客户端，客户端汇集各个RegionServer上的操作结果，做最后的处理，如计算各个RegionServer返回结果的和等。我这里的代码如下，仅做一个示例，有很多不完善的地方：

代码中声明了一个RPCCallback，通过RPCCallBack的get方法，就可以得到在RegionServer执行调用获得的结果，各个调用的结果会返回到Map容器中，然后通过对Map容器的迭代，客户端即可获得各个RegionServer上的执行结果，从而进行自定义的计算。

package com.uestc.coprocessor;

import java.io.IOException;

import java.lang.reflect.Field;

import java.util.LinkedHashMap;

import java.util.Map;

import org.apache.hadoop.hbase.client.Table;

import org.apache.hadoop.hbase.client.coprocessor.Batch;

import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;

import com.google.protobuf.ServiceException;

import com.uestc.util.JsonUtil;

import com.uestc.util.Log;

public class MultiColumnAggregateClient {

public static class CountInfo {

public long qiniuCount;

public long qiniu2Count;

public long jinshanCount;

public long otherCount;

@Override

public String toString() {

String rString = String.format("qiniuCount:%d;Qiniu2Count:%d;JinshanCount:%d;OtherCount:%d",

this.qiniuCount, this.qiniu2Count, this.jinshanCount, this.otherCount);

return rString;

}

/**

* 生成jsonString

* 

* @return

* @throws IllegalArgumentException

* @throws IllegalAccessException

*/

public String toJsonString() throws IllegalArgumentException, IllegalAccessException {

Field[] fields = this.getClass().getFields();

Map<String, Object> map = new LinkedHashMap<>();

for (Field field : fields) {

map.put(field.getName(), field.get(this));

}

return JsonUtil.createJsonString(map);

}

}

public CountInfo getSeveralCounts(Table table, String columnNames) throws ServiceException, Throwable {

final JAggregateProtocol.CountRequest request = JAggregateProtocol.CountRequest.newBuilder()

.setColumns(columnNames).build();

Map<byte[], CountInfo> map = table.coprocessorService(JAggregateProtocol.RowCountService.class, null, null,

new Batch.Call<JAggregateProtocol.RowCountService, CountInfo>() {

@Override

public CountInfo call(JAggregateProtocol.RowCountService aggregate) throws IOException {

BlockingRpcCallback<JAggregateProtocol.CountResponse> rpcCallback = new BlockingRpcCallback<>();

aggregate.getSizes(null, request, rpcCallback);

JAggregateProtocol.CountResponse response = rpcCallback.get();

CountInfo countInfo = new CountInfo();

countInfo.qiniuCount = response.getQiniuCount();

countInfo.qiniu2Count = response.getQiniu2Count();

countInfo.jinshanCount = response.getJinshanCount();

countInfo.otherCount = response.getOtherCloud();

return countInfo;

}

});

CountInfo result = new CountInfo();

for (CountInfo countInfo : map.values()) {

result.qiniuCount += countInfo.qiniuCount;

result.qiniu2Count += countInfo.qiniu2Count;

result.jinshanCount += countInfo.jinshanCount;

result.otherCount += countInfo.otherCount;

}

Log.logger.info(result);

return result;

}

}

这里分成两步

（1）、可用Eclipse把MultiColumnAggregation类文件（即需要在RegionServer上运行的类）打成jar包，然后把它分发到各个HBase节点上，并把它们放到HBase目录的lib子目录下，这是最简单的方法；或者也可以去配置CLASSPATH、HBASE_CLASSPATH之类的，保证HBase能访问到我们的jar文件。

（2）、编辑HBASE_HOME/conf/hbase-site.xml文件，配置Coprocessor类

若有多个Coprocessor，则用逗号分隔，value标签中的第一个Coprocessor是HBase自带的，第二个是我们自己自定义的。

<property>

  <name>hbase.coprocessor.region.classes</name>

  <value>org.apache.hadoop.hbase.coprocessor.AggregateImplementation,com.uestc.coprocessor.MultiColumnAggregation</value>

</property>

可以这样来调用刚才我们自己写的Coprocessor：

这里的MTable是我自己封装的Table类，它的getTable方法即返回Table类的实例。

public MultiColumnAggregateClient.CountInfo sumUpSizes() throws IOException {

MultiColumnAggregateClient client = new MultiColumnAggregateClient();

MultiColumnAggregateClient.CountInfo info = null;

MTable mTable = null;

try {

mTable = new MTable("file");

info = client.getSeveralCounts(mTable.getTable(), "Qiniu;Qiniu2;Jinshan");

} catch (Throwable e) {

Log.logException(e);

} finally {

if (mTable != null)

mTable.close();

}

return info;

}

以上的代码因为项目上赶时间临时写的，有很多不完善的地方，但是也完成了需求，通过这个自定义的Coprocessor，就可以一次调用同时统计多个列的的和了，极大的提高了计算的效率。