码迷,mamicode.com
首页 > 其他好文 > 详细

HBase协处理器实战

时间:2016-11-14 07:48:08      阅读:188      评论:0      收藏:0      [点我收藏+]

标签:ons   知识   客户端   ret   nts   convert   region   特性   start   

一、协处理器介绍

定义:HBase可以让用户的部分逻辑在数据存放端及hbase服务端进行计算的机制(框架)。协处理器允许用户在hbase服务端上运行自己的代码。

 

分类:系统协处理器、表协处理器 

observer,相当于关系型数据库里面的触发器 

endpoint,类似于存储过程 

二、OBserver

观察者的设计意图是允许用户通过插入代码来重载协处理器框架的upcall方法,而具体的事件触发的callback方法由HBase的核心代码来执行。可以分为RegionObserver、RegionServerObserver、MasterObserver、WalObserver 

RegionObserver              提供客户端的数据操纵事件钩子: Get、 Put、 Delete、Scan等

RegionServerObserver     专门处理RegionServer上的一些事件

MasterObserver              提供DDL-类型的操作钩子。如创建、删除、修改数据表等

WalObserver                  提供WAL相关操作

 这些接口可以同时使用在同一个地方,按照不同优先级顺序执行.用户可以任意基于协处理器实现复杂的HBase功能层。HBase有很多种事件可以触发观察者方法,这些事件与方法从HBase0.92版本起,都会集成在HBase API中。不过这些API可能会由于各种原因有所

改动,不同版本的接口改动比较大。RegionObserver工作原理如下图所示。

技术分享

三、Endpoint

终端是动态RPC插件的接口,它的实现代码被安装在服务器端,从而能够通过HBase RPC唤醒。客户端类库提供了非常方便的方法来调用这些动态接口,它们可以在任意时候调用一个终端,它们的实现代码会被目标region远程执行,结果会返回到终端。用户可以结合使

用这些强大的插件接口,为HBase添加全新的特性。

技术分享

 

endpoint服务端编写

1.创建endpoint.proto文件,生成java文件

  1. option java_package = "edu.endpoint";  
  2. option java_outer_classname = "Sum";  
  3. option java_generic_services = true;  
  4. option java_generate_equals_and_hash = true;  
  5. option optimize_for = SPEED;  
  6. message SumRequest {  
  7.     required string family = 1;  
  8.     required string column = 2;  
  9. }  
  10. message SumResponse {  
  11.     required int64 sum = 1 [default = 0];  
  12. }  
  13.   
  14. service SumService {  
  15.     rpc getSum(SumRequest)  
  16.         returns (SumResponse);  
  17. }  

2.将.proto文件生成Java代码。 

protoc endpoint.proto –java_out=./ 

将生成的.java文件拷贝到eclipse对应的java项目源代码包文件下。添加hbase lib下所有依赖库文件。 

3.编写服务端SumEndpoint.java类

  1. package edu.endpoint;  
  2.   
  3. import java.io.IOException;  
  4. import java.util.ArrayList;  
  5. import java.util.List;  
  6.   
  7. import org.apache.hadoop.hbase.Coprocessor;  
  8. import org.apache.hadoop.hbase.CoprocessorEnvironment;  
  9. import org.apache.hadoop.hbase.client.Scan;  
  10. import org.apache.hadoop.hbase.coprocessor.CoprocessorException;  
  11. import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;  
  12. import org.apache.hadoop.hbase.regionserver.InternalScanner;  
  13. import org.apache.hadoop.hbase.util.Bytes;  
  14.   
  15. import com.google.protobuf.RpcCallback;  
  16. import com.google.protobuf.RpcController;  
  17. import com.google.protobuf.Service;  
  18.   
  19. import edu.endpoint.Sum.SumRequest;  
  20. import edu.endpoint.Sum.SumResponse;  
  21. import edu.endpoint.Sum.SumService;  
  22.   
  23. public class SumEndPoint extends SumService implements Coprocessor,CoprocessorService{  
  24.     private RegionCoprocessorEnvironment env;  
  25.   
  26.     public void getSum(RpcController controller,SumRequest request,RpcCallback<SumResponse> done) throws IOException{  
  27.         Scan scan = new Scan();  
  28.         scan.addFamily(Bytes.toBytes(request.getFamily()));  
  29.         scan.addColumn(Bytes.toBytes(request.getFamily()), Bytes.toBytes(request.getColumn()));  
  30.         SumResponse response = null;  
  31.         InternalScanner scanner = null;  
  32.         try{  
  33.             scanner = env.getRegion().getScanner(scan);  
  34.             List<Cell> results = new ArrayList<Cell>();  
  35.             boolean hasMore = false;  
  36.             Long sum = 0;  
  37.             do {  
  38.                 hasMore = scanner.next(results);  
  39.                 for(Cell cell:results){  
  40.                     sum += Long.parseLong(new String(CellUtil.cloneValue(cell)));  
  41.                 }  
  42.                 results.clear();  
  43.             } while (hasMore);  
  44.             response = SumResponse.newBuilder().setSum(sum).build();  
  45.         }catch (IOException e) {  
  46.             ResponseConverter.setControllerException(controller,e);  
  47.         }finally {  
  48.                 if (scanner!=null) {  
  49.                     try {  
  50.                         scanner.close();  
  51.                     } catch (IOException e) {  
  52.                         // TODO Auto-generated catch block  
  53.                         e.printStackTrace();  
  54.                     }     
  55.                 }  
  56.                   
  57.         }  
  58.         done.run(response);  
  59.     }  
  60.     public Service getService(){  
  61.         return this;  
  62.     }  
  63.     @Override  
  64.     public void start(CoprocessorEnvironment env) throws IOException {  
  65.         if (env instanceof RegionCoprocessorEnvironment) {  
  66.             this.env =(RegionCoprocessorEnvironment) env;  
  67.         }else {  
  68.             throw new CoprocessorException("no load region");  
  69.         }  
  70.     }  
  71.     @Override  
  72.     public void stop(CoprocessorEnvironment arg0) throws IOException {  
  73.     }  
  74. }  

 

 


 

HBase协处理器实战

标签:ons   知识   客户端   ret   nts   convert   region   特性   start   

原文地址:http://www.cnblogs.com/fangdai/p/6060569.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!