阿里云
发表主题 回复主题
  • 510阅读
  • 1回复

HBase-1.0.1.1 编写Coprocessor【分享】

级别: 新人
发帖
0
云币
-40




这几天为了满足业务需求,自己写了个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的介绍网上有很多,这里就不再介绍了。




1.首先需要下载protobuf的解析器protobuf-2.5.0.tar.gz,然后,按照如下的方式安装解析器。

[Bash shell] 纯文本查看 复制代码

tar -zxvf protobuf-2.5.0.tar.gz
./configure --prefix=/home/cyber_space/protobuf
make
make install





2.编写.proto文件

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




[Bash shell] 纯文本查看 复制代码

/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文件的路径在如下的路径中,

[Bash shell] 纯文本查看 复制代码

~/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表示的是变量的优先级。


[Bash shell] 纯文本查看 复制代码

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代码,使用的命令如下:

[Bash shell] 纯文本查看 复制代码

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





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



3.接下来要写一个类,这个类将在RegionServer端运行,然后被客户端通过RPC框架调用

这个实现类需要继承我们用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即可。



[Bash shell] 纯文本查看 复制代码

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);
}

}





4.再编写一个客户端类来调用我们刚才写的代码。

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


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



[Bash shell] 纯文本查看 复制代码

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;
}
}





5.部署我们的Coprocessor

这里分成两步


(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自带的,第二个是我们自己自定义的。





[Bash shell] 纯文本查看 复制代码

<property>
<name>hbase.coprocessor.region.classes</name>
<value>org.apache.hadoop.hbase.coprocessor.AggregateImplementation,com.uestc.coprocessor.MultiColumnAggregation</value>
</property>




6.编写调用代码

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


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




[Bash shell] 纯文本查看 复制代码

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,就可以一次调用同时统计多个列的的和了,极大的提高了计算的效率。




级别: *
发帖
*
云币
*
file的表结构如果能放上来就更好了.
发表主题 回复主题
« 返回列表
«12345678910»
共10页
上一主题下一主题

限100 字节
如果您在写长篇帖子又不马上发表,建议存为草稿
 
验证问题: ECS是阿里云提供的什么服务? 正确答案:云服务器
上一个 下一个
      ×
      全新阿里云开发者社区, 去探索开发者的新世界吧!
      一站式的体验,更多的精彩!
      通过下面领域大门,一起探索新的技术世界吧~ (点击图标进入)

      版权声明

      开发者论坛为你提供“HBase-1.0.1.1 编写Coprocessor【分享】”的内容,论坛中还有更多关于 getserviceCoprocessorresponse类写的文件import编写纯 的内容供你使用,该内容是网友上传,与开发者论坛无关,如果需要删除请联系zixun-group@service.aliyun.com,工作人员会在5个工作日内回复您。