hbase0.98 coprocessor Endpoint 之 HelloWorld

時間 2019-11-06

標籤 hbase0.98 hbase coprocessor endpoint helloworld 欄目 Hadoop 简体版

原文原文鏈接

###介紹： 源博地址: http://www.cockybook.com/?p=35html

HBase做爲列族數據庫最常常被人詬病的特性包括：沒法輕易創建「二級索引」，難以執行求和、計數、排序等操做。好比，在舊版本的(<0.92)Hbase中，統計數據表的總行數，須要使用Counter方法，執行一次MapReduce Job才能獲得。雖然HBase在數據存儲層中集成了MapReduce，可以有效用於數據表的分佈式計算。然而在不少狀況下，作一些簡單的相加或者聚合計算的時候，若是直接將計算過程放置在server端，可以減小通信開銷，從而得到很好的性能提高。因而，HBase在0.92以後引入了協處理器(coprocessors)，實現一些激動人心的新特性：可以輕易創建二次索引、複雜過濾器(謂詞下推)以及訪問控制等。 HBase協處理器的靈感來自於Jeff Dean 09年的演講( P66-67)。java

####hbase coprocessor 大類分爲兩種coprocessor分別是：web

RegionObserver ：它是一種相似於傳統數據庫的觸發器，提供了鉤子函數：Get、Put、Delete、Scan等。

Endpoint：是一個遠程rpc調用，相似於webservice形式調用，但他不適用xml，而是使用的序列化框架是protobuf（序列化後數據更小），本文將介紹此種Coprocessor.

Endpoint 容許您定義本身的動態RPC協議,用於客戶端與region servers通信。Coprocessor 與region server在相同的進程空間中，所以您能夠在region端定義本身的方法（endpoint），將計算放到region端，減小網絡開銷，經常使用於提高hbase的功能，如：count，sum等。shell

###個人環境數據庫

hadoop ： 2.2
hbase-hadoop2 ：0.98+
JDK：1.6 ##這裏必需要1.6 要否則會出現不能加載jar包的現象。
操做系統：CentOS 6.4

###編寫代碼apache

首先你須要利用protobuf（網上本身搜google維護的目前發展到2.5版本）工具成一個HelloWorld 序列化對象。api

####HelloWorld.proto網絡

option java_package = "com.gzhdi.coprocessor.generated";
option java_outer_classname = "ServerHelloworld";
option java_generic_services = true;
option java_generate_equals_and_hash = true;
option optimize_for = SPEED;

message HelloRequest {
  required bytes askWord = 10;
}

message HelloResponse {
  required bytes retWord = 10;
}

message AskRequest {
  required bytes ask = 100;
}

message AnsResponse {
  required bytes ans = 100;
}

service HelloWorld {
  rpc sendHello(HelloRequest)
    returns (HelloResponse);

  rpc question(AskRequest)
    returns (AnsResponse);
}

使用命令生成代碼，並拷貝到你的工程裏邊去,個人文件在工程下面放着呢，直接生成到工程裏邊。這段代碼就會生成一個HelloWorld.java文件.框架
```
protoc.exe  --java_out=../src HelloWorld.proto
```

編寫主要代碼分佈式

####server端代碼

package com.gzhdi.copocessor;

import java.io.IOException;

import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;

import com.google.protobuf.ByteString;
import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
import com.google.protobuf.Service;
import com.gzhdi.coprocessor.generated.ServerHelloworld;
import com.gzhdi.coprocessor.generated.ServerHelloworld.AnsResponse;
import com.gzhdi.coprocessor.generated.ServerHelloworld.AskRequest;
import com.gzhdi.coprocessor.generated.ServerHelloworld.HelloRequest;
import com.gzhdi.coprocessor.generated.ServerHelloworld.HelloResponse;

public class HelloWorldEndPoint  extends ServerHelloworld.HelloWorld implements Coprocessor,CoprocessorService{

	private RegionCoprocessorEnvironment env; 

	@Override
	public void sendHello(RpcController controller, HelloRequest request,
			RpcCallback<HelloResponse> done) {
		System.out.println("request HelloRequest:"+request.getAskWord());
		HelloResponse resp=HelloResponse.newBuilder().setRetWord(ByteString.copyFromUtf8("hello world!!!")).build();

		done.run(resp);
	}

	@Override
	public void question(RpcController controller, AskRequest request,
			RpcCallback<AnsResponse> done) {
		System.out.println("request question:"+request.getAsk());
		AnsResponse resp=AnsResponse.newBuilder().setAns(ByteString.copyFromUtf8("helloworld,"+request.getAsk().toStringUtf8())).build();
		done.run(resp);
	}

	@Override
	public Service getService() {
		return this;
	}

	@Override
	public void start(CoprocessorEnvironment env) throws IOException {
		if (env instanceof RegionCoprocessorEnvironment) {  
		      this.env = (RegionCoprocessorEnvironment)env;  
		    } else {  
		      throw new CoprocessorException("Must be loaded on a table region!");  
		    }  
	}

	@Override
	public void stop(CoprocessorEnvironment env) throws IOException {

	}
}

####client 端代碼

package com.gzhdi.copocessor;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;
import org.apache.hadoop.hbase.ipc.ServerRpcController;

import com.google.protobuf.ByteString;
import com.google.protobuf.ServiceException;
import com.gzhdi.coprocessor.generated.ServerHelloworld.AnsResponse;
import com.gzhdi.coprocessor.generated.ServerHelloworld.AskRequest;
import com.gzhdi.coprocessor.generated.ServerHelloworld.HelloWorld;

public class HelloWorldClient {

	public static void main(String[] args) throws ServiceException, Throwable {
		myclient();
	}
//若是你沒有寫好本身的例子能夠跑跑hbase自帶的小例子
//	private static void example1() throws IOException, ServiceException,
//			Throwable {
//		System.out.println("begin.....");  
//        long begin_time=System.currentTimeMillis();  
//       Configuration config=HBaseConfiguration.create();  
////     String master_ip="192.168.150.128";  
//       String master_ip="10.10.113.211";  
//       String zk_ip="10.10.113.211";  
//       String table_name="t1";  
//       config.set("hbase.zookeeper.property.clientPort", "2181");   
//       config.set("hbase.zookeeper.quorum", zk_ip);   
//       config.set("hbase.master", master_ip+":600000");  
//       
//       HTable table = new HTable(config, table_name);
//       final ExampleProtos.CountRequest request = ExampleProtos.CountRequest.getDefaultInstance();
//       Map results = table.coprocessorService(
//           ExampleProtos.RowCountService.class, // the protocol interface we're invoking
//           null, null,                          // start and end row keys
//           
//           new Batch.Call() {
//        	   
//               public Long call(Object counter) throws IOException {
//                 BlockingRpcCallback<ExampleProtos.CountResponse> rpcCallback =
//                     new BlockingRpcCallback();
//                 ((ExampleProtos.RowCountService)counter).getRowCount(null, request, rpcCallback);
//                 ExampleProtos.CountResponse response = rpcCallback.get();
//                 System.out.println("count :::::"+response.getCount());
//                 return response.hasCount() ? response.getCount() : 0;
//               }
//			
//           });
//	}

	public static void myclient(){
		// TODO Auto-generated method stub
				System.out.println("begin.....");  
		        long begin_time=System.currentTimeMillis();  
		       Configuration config=HBaseConfiguration.create();  
//		     String master_ip="192.168.150.128";  
		       String master_ip="10.10.113.211";  
		       String zk_ip="10.10.113.211";  
		       String table_name="t1";  
		       config.set("hbase.zookeeper.property.clientPort", "2181");   
		       config.set("hbase.zookeeper.quorum", zk_ip);   
		       config.set("hbase.master", master_ip+":600000");  

		       final AskRequest req=AskRequest.newBuilder().setAsk(ByteString.copyFromUtf8("hello")).build();
		       AnsResponse resp=null;
		       try {
				HTable table=new HTable(config,table_name);
				Map<byte[], ByteString> re=table.coprocessorService(HelloWorld.class, null, null, new Batch.Call<HelloWorld, ByteString>() {

					@Override
					public ByteString call(HelloWorld instance) throws IOException {
						ServerRpcController controller = new ServerRpcController();
						BlockingRpcCallback<AnsResponse> rpccall=new BlockingRpcCallback<AnsResponse>();
						instance.question(controller, req, rpccall);
						AnsResponse resp=rpccall.get();


						//result
						System.out.println("resp:"+ resp.getAns().toStringUtf8());

						return resp.getAns();
					}

				});
			} catch (IOException e) {
				e.printStackTrace();
			} catch (ServiceException e) {
				e.printStackTrace();
			} catch (Throwable e) {
				e.printStackTrace();
			}  
	}

}

利用jdk 1.6打包（切記jdk1.6，由於hbase用1.6打包的）導出hellworld.jar 包名隨便起。

###部署

將包helloworld.jar 放在 %HBASE_HOME/lib/ 下就能夠了。
從新啓動hbase

驗證

[root@hdp22 ~ Desktop]# hbase shell
hbase(main):001:0> import com.gzhdi.copocessor.HelloWorldEndPoint
=> Java::ComGzhdiCopocessor::HelloWorldEndPoint    //若是打印出這句話就說明包已經加載完畢

向指定表添加endpoint

hbase(main):002:0> create 't1','f1'
0 row(s) in 6.5290 seconds
=> Hbase::Table - t1   //建立表t1

hbase(main):003:0> alter 't1','coprocessor'=>'|com.gzhdi.copocessor.HelloWorldEndPoint|1001|'
Updating all regions with the new schema...
0/1 regions updated.
1/1 regions updated.
Done.
0 row(s) in 2.5960 seconds

hbase(main):005:0> describe 't1'
DESCRIPTION                                                                                                              ENABLED                                                          
 't1', {TABLE_ATTRIBUTES => {coprocessor$1 => '|com.gzhdi.copocessor.HelloWorldEndPoint|1001|'}, {NAME => 'f1', DATA_BLO true                                                             
 CK_ENCODING => 'NONE', BLOOMFILTER => 'ROW', REPLICATION_SCOPE => '0', VERSIONS => '1', COMPRESSION => 'NONE', MIN_VERS                                                                  
 IONS => '0', TTL => '2147483647', KEEP_DELETED_CELLS => 'false', BLOCKSIZE => '65536', IN_MEMORY => 'false', BLOCKCACHE                                                                  
  => 'true'}                                                                                                                                                                              
1 row(s) in 0.0940 seconds

//OK 成功了

###調用如今就可使用你的客戶端代碼調用該服務了，須要制定zookeeper地址和表名（由於服務是針對表的）。

###參考資料 http://blog.csdn.net/yangzongzhao/article/details/24306775 https://blogs.apache.org/hbase/entry/coprocessor_introduction#comment-1337006507000 http://blog.csdn.net/hljlzc2007/article/details/12652243 http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/client/coprocessor/package-summary.html http://hbase.apache.org/apidocs/org/apache/hadoop/hbase/coprocessor/package-summary.html