技术篇-HBase Coprocessor 的实现与应用-阿里云开发者社区

本次分享的内容主要分为以下五点:

Coprocessor 简介
Endpoint 服务端实现
Endpoint 客户端实现
Observer 实现二级索引
Coprocessor 应用场景

_2019_01_11_1_54_38

1. Coprocessor 简介

HBase 协处理器的灵感来自于 Jeff Dean 09 年的演讲，根据该演讲实现类似于 Bigtable 的协处理器，包括以下特性:每个表服务器的任意子表都可以运行代码客户端的高层调用接口(客户端能够直接访问数据表的行地址，多行读写会自动分片成多个并行的 RPC 调用)，提供一个非常灵活的、可用于建立分布式服务的数据模型，能够自动化扩展、负载均衡、应用请求路由。HBase 的协处理器灵感来自 Bigtable，但是实现细节不尽相同。HBase 建立框架为用户提供类库和运行时环境，使得代码能够在 HBase Region Server 和 Master 上面进行处理。

_2019_01_11_1_55_41

(1)实现目的

HBase 无法轻易建立“二级索引”;
执行求和、计数、排序等操作比较困难，必须通过 MapReduce/Spark 实现，
对于简单的统计或聚合计算，可能会因为网络与 IO 开销大而带来性能问题。

(2)灵感来源

灵感来源于 Bigtable 的协处理器，包含如下特性:

每个表服务器的任意子表都可以运行代码;
客户端能够直接访问数据表的行，多行读写会自动分片成多个并行的 RPC 调
用。

(3)提供接口

RegionObserver:提供客户端的数据操纵事件钩子:Get、Put、Delete、Scan 等;
WALObserver:提供 WAL 相关操作钩子;
MasterObserver:提供 DDL-类型的操作钩子。如创建、删除、修改数据表等;
Endpoint:终端是动态 RPC 插件的接口，它的实现代码被安装在服务器端，能够通过 HBase RPC 调用唤醒。

(4)应用范围

通过使用 RegionObserver 接口可以实现二级索引的创建和维护;
通过使用 Endpoint 接口，在对数据进行简单排序和 sum，count 等统计操作时，能够极大提高性能。
本文将通过具体实例来演示两种协处理器的开发方法的详细实现过程。

2. Endpoint 服务端实现

在传统关系型数据库里面，可以随时的对某列进行求和 sum，但是目前 HBase 目前所提供的接口，直接求和是比较困难的，所以先编写好服务端代码，并加载到对应的 Table 上，加载协处理器有几种方法，可以通过 HTableDescriptor 的 addCoprocessor 方法直接加载，同理也可以通过 removeCoprocessor 方法卸载协处理器。

Endpoint 协处理器类似传统数据库的存储过程，客户端调用 Endpoint 协处理器执行一段 Server 端代码，并将 Server 端代码的结果返回给 Client 进一步处理，最常见的用法就是进行聚合操作。举个例子说明:如果没有协处理器，当用户需要找出一张表中的最大数据即 max 聚合操作，必须进行全表扫描，客户端代码遍历扫描结果并执行求 max 操作，这样的方法无法利用底层集群的并发能力，而将所有计算都集中到 Client 端统一执行，效率非常低。但是使用 Coprocessor，用户将求 max 的代码部署到 HBase Server 端，HBase 将利用底层 Cluster 的多个节点并行执行求 max 的操作即在每个 Region 范围内执行求最大值逻辑，将每个 Region 的最大值在 Region Server 端计算出，仅仅将该 max 值返回给客户端。客户端进一步将多个 Region 的 max 进一步处理而找到其中的 max，这样整体执行效率提高很多。但是一定要注意的是 Coprocessor 一定要写正确，否则导致 RegionServer 宕机。

_2019_01_11_2_53_52

2.1 Protobuf 定义

如前所述，客户端和服务端之间需要进行 RPC 通信，所以两者间需要确定接口，当前版本的 HBase 的协处理器是通过 Google Protobuf 协议来实现数据交换的，所以需要通过 Protobuf 来定义接口。
如下所示:


option java_package = "com.my.hbase.protobuf.generated"; 

option java_outer_classname = "AggregateProtos";

option java_generic_services = true;

option java_generate_equals_and_hash = true; 

option optimize_for = SPEED;

import "Client.proto";


message AggregateRequest {
       required string interpreter_class_name = 1; required Scan scan = 2;
       optional bytes interpreter_specific_bytes = 3;
}
       message AggregateResponse { repeated bytes first_part = 1; 
       optional bytes second_part = 2;
}


service AggregateService {
     rpc GetMax (AggregateRequest) returns (AggregateResponse);
     rpc GetMin (AggregateRequest) returns (AggregateResponse);
     rpc GetSum (AggregateRequest) returns (AggregateResponse); 
     rpc GetRowNum (AggregateRequest) returns (AggregateResponse); 
     rpc GetAvg (AggregateRequest) returns (AggregateResponse); 
     rpc GetStd (AggregateRequest) returns (AggregateResponse); rpc GetMedian (AggregateRequest) returns (AggregateResponse);
}

可以看到这里定义 7 个聚合服务 RPC，名字分别叫做 GetMax、GetMin、GetSum 等，本文通过 GetSum 进行举例，其他的聚合 RPC 也是类似的内部实现。RPC 有一个入口参数，用消息 AggregateRequest 表示;RPC 的返回值用消息 AggregateResponse 表示。Service 是一个抽象概念，RPC 的 Server 端可以看作一个用来提供服务的 Service。在 HBase Coprocessor 中 Service 就是 Server 端需要提供的 Endpoint Coprocessor 服务，主要用来给 HBase 的 Client 提供服务。 AggregateService.java 是由 Protobuf 软件通过终端命令“protoc filename.proto-- java_out=OUT_DIR”自动生成的，其作用是将.proto 文件定义的消息结构以及服务转换成对应接口的 RPC 实现，其中包括如何构建 request 消息和 response 响应以及消息包含的内容的处理方式，并且将 AggregateService 包装成一个抽象类，具体的服务以类的方法的形式提供。

AggregateService.java 定义 Client 端与 Server 端通信的协议，代码中包含请求信息结构 AggregateRequest、响应信息结构 AggregateResponse 、提供的服务种类 AggregateService ，其中 AggregateRequest 中的 interpreter_class_name 指的是 column interpreter 的类名，此类的作用在于将数据格式从存储类型解析成所需类型。 AggregateService.java 由于代码太长，在这里就不贴出来了。

下面我们来讲一下服务端的架构:

首先，Endpoint Coprocessor 是一个 Protobuf Service 的实现，因此需要它必须继承某个 ProtobufService。我们在前面已经通过 proto 文件定义 Service，命名为 AggregateService，因此 Server 端代码需要重载该类，其次作为 HBase 的协处理器，Endpoint 还必须实现 HBase 定义的协处理器协议，用 Java 的接口来定义。具体来说就是 CoprocessorService 和 Coprocessor，这些 HBase 接口负责将协处理器和 HBase 的 RegionServer 等实例联系起来以便协同工作。Coprocessor 接口定义两个接口函数:start 和 stop。

加载 Coprocessor 之后 Region 打开的时候被 RegionServer 自动加载，并会调用器 start 接口完成初始化工作。一般情况该接口函数仅仅需要将协处理器的运行上下文环境变量 CoprocessorEnvironment 保存到本地即可。

CoprocessorEnvironment 保存协处理器的运行环境，每个协处理器都是在一个 RegionServer 进程内运行并隶属于某个 Region。通过该变量获取 Region 的实例等 HBase 运行时环境对象。

Coprocessor 接口还定义 stop()接口函数，该函数在 Region 被关闭时调用，用来进行协处理器的清理工作。本文里我们没有进行任何清理工作，因此该函数什么也不干。

我们的协处理器还需要实现 CoprocessorService 接口。该接口仅仅定义一个接口函数 getService()。我们仅需要将本实例返回即可。HBase 的 Region Server 在接收到客户端的调用请求时，将调用该接口获取实现 RPCService 的实例，因此本函数一般情况下就是返回自身实例即可。

完成以上三个接口函数之后，Endpoint 的框架代码就已完成。每个 Endpoint 协处理器都必须实现这些框架代码而且写法雷同。

_2019_01_11_3_09_39

Server 端的代码就是一个 Protobuf RPC 的 Service 实现，即通过 Protobuf 提供的某种服务。其开发内容主要包括:

实现 Coprocessor 的基本框架代码
实现服务的 RPC 具体代码

2.2 Endpoint 协处理的基本框架

Endpoint 是一个Server端Service的具体实现，其实现有一些框架代码，这些框架代码与具体的业务需求逻辑无关。仅仅是为了和 HBase 运行时环境协同工作而必须遵循和完成的一些粘合代码。因此多数情况下仅仅需要从一个例子程序拷贝过来并进行命名修改即可。不过我们还是完整地对这些粘合代码进行粗略的讲解以便更好地理解代码。


public Service getService() { 
return this; } 
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!"); } 
} 
public void stop(CoprocessorEnvironment env) throws IOException { }

下面我们来讲一下服务端的架构:

我们的协处理器还需要实现 CoprocessorService 接口。该接口仅仅定义一个接口函数 getService()。我们仅需要将本实例返回即可。
HBase 的 Region Server 在接收到客户端的调用请求时，将调用该接口获取实现 RPCService 的实例，因此本函数一般情况下就是返回自身实例即可。

完成以上三个接口函数之后，Endpoint 的框架代码就已完成。每个 Endpoint 协处理器都必须实现这些框架代码而且写法雷同。

_2019_01_11_3_17_16

Server 端的代码就是一个 Protobuf RPC 的 Service 实现，即通过 Protobuf 提供的某种服务。其开发内容主要包括:

实现 Coprocessor 的基本框架代码
实现服务的 RPC 具体代码

2.2 Endpoint 协处理的基本框架


public Service getService() { 
return this; } 
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!"); } 
} 
public void stop(CoprocessorEnvironment env) throws IOException { }

Endpoint 协处理器真正的业务代码都在每一个 RPC 函数的具体实现中。

在本文中，我们的 Endpoint 协处理器仅提供一个 RPC 函数即 getSUM。我将分别介绍编写该函数的几个主要工作:了解函数的定义，参数列表;处理入口参数; 实现业务逻辑;设置返回参数。


public void getSum(RpcController controller, AggregateRequest request, RpcCal lbackdone) { 
AggregateResponse response = null; 
RegionScanner scanner = null; 
long sum = 0L;
 try { 
ColumnInterpreter ignored = this.constructColumnInterpreterFromRequ est(request); 
Object sumVal = null; 
Scan scan = ProtobufUtil.toScan(request.getScan()); 
scanner = this.env.getRegion().getScanner(scan); byte[] colFamily = scan.getFamilies()[0]; 
lFamily); 
NavigableSet qualifiers = (NavigableSet) scan.getFamilyMap().get(co 
byte[] qualifier = null; 
if (qualifiers != null && !qualifiers.isEmpty()) { 
qualifier = (byte[]) qualifiers.pollFirst();
 } 
ArrayList results = new ArrayList(); 
boolean hasMoreRows = false; 
do{ hasMoreRows = scanner.next(results); 
int listSize = results.size(); 
for(inti=0;
i<listSize;++i){ 
p)); 
//取出列值 Object temp = ignored.getValue(colFamily, qualifier, 
(Cell) results.get(i)); 
if (temp != null) { 
sumVal = ignored.add(sumVal, ignored.castToReturnType(tem 
}


} 
results.clear(); 
} while (hasMoreRows); 
if (sumVal != null) { response = AggregateResponse.newBuilder().addFirstPart(               ignored.getProtoForPromotedType(sumVal).toByteString()).
build(); 
} 
} catch (IOException var27) { ResponseConverter.setControllerException(controller, var27); 
} finally {
 if (scanner != null) { 
try { scanner.close(); 
} catch (IOException var26) { ; 
} } 
} 
log.debug("Sum from this region is " + this.env.getRegion().getRegionInfo().getRegionNameAsString() + 
":"+sum); 
done.run(response); 
} 

 /**

Endpoint 类比于数据库的存储过程，其触发服务端的基于 Region 的同步运行再将各个结果在客户端搜集后归并计算。特点类似于传统的 MapReduce 框架，服务端 Map 客户端 Reduce。

3. Endpoint 客户端实现

HBase 提供客户端 Java 包 org.apache.hadoop.hbase.client.HTable，提供以下三种方法来调用协处理器提供的服务:

coprocessorService(byte[])
coprocessorService(Class, byte[], byte[],Batch.Call),
coprocessorService(Class, byte[], byte[],Batch.Call,
```
Batch.Callback)
```

该方法采用 rowkey 指定 Region。这是因为 HBase 客户端很少会直接操作 Region，一般不需要知道 Region 的名字;况且在 HBase 中 Region 名会随时改变，所以用 rowkey 来指定 Region 是最合理的方式。使用 rowkey 可以指定唯一的一个 Region，如果给定的 Rowkey 并不存在，只要在某个 Region 的 rowkey 范围内依然用来指定该 Region。比如 Region 1 处理[row1, row100]这个区间内的数据，则 rowkey=row1 就由 Region 1 来负责处理，换句话说我们可以用 row1 来指定 Region 1，无论 rowkey 等于”row1”的记录是否存在。CoprocessorService 方法返回类型为 CoprocessorRpcChannel 的对象，该 RPC 通道连接到由 rowkey 指定的 Region 上面，通过此通道可以调用该 Region 上面部署的协处理器 RPC。

_2019_01_11_3_36_05

有时候客户端需要调用多个 Region 上的同一个协处理器，比如需要统计整个 Table 的 sum，在这种情况下，需要所有的 Region 都参与进来，分别统计自身 Region 内部的 sum 并返回客户端，最终客户端将所有 Region 的返回结果汇总，就可以得到整张表的 sum。

这意味着该客户端同时和多个 Region 进行批处理交互。一个可行的方法是，收集每个 Region 的 startkey，然后循环调用第一种 coprocessorService 方法:用每一个 Region 的 startkey 作为入口参数，获得 RPC 通道创建 stub 对象，进而逐一调用每个 Region 上的协处理器 RPC。这种做法需要写很多的代码，为此 HBase 提供两种更加简单的 coprocessorService 方法来处理多个 Region 的协处理器调用。先来看第一种方法 coprocessorService(Class, byte[],byte[],Batch.Call)

该方法有 4 个入口参数。第一个参数是实现 RPC 的 Service 类，即前文中的 AggregateService 类。通过它，HBase 就可以找到相应的部署在 Region 上的协处理器，一个 Region 上可以部署多个协处理器，客户端必须通过指定 Service 类来区分究竟需要调用哪个协处理器提供的服务。

要调用哪些 Region 上的服务则由 startkey 和 endkey 来确定，通过 rowkey 范围即可确定多个 Region。为此，coprocessorService 方法的第二个和第三个参数分别是 startkey 和 endkey，凡是落在[startkey，endkey]区间内的 Region 都会参与本次调用。

第四个参数是接口类 Batch.Call。它定义了如何调用协处理器，用户通过重载该接口的 call()方法来实现客户端的逻辑。在 call()方法内，可以调用 RPC，并对返回值进行任意处理。即前文代码清单 1 中所做的事情。coprocessorService 将负责对每个 Region 调用这个 call()方法。

coprocessorService 方法的返回值是一个 Map 类型的集合。该集合的 key 是 Region 名字，value 是 Batch.Call.call 方法的返回值。该集合可以看作是所有 Region 的协处理器 RPC 返回的结果集。客户端代码可以遍历该集合对所有的结果进行汇总处理。

这种 coprocessorService 方法的大体工作流程如下。首先它分析 startkey 和 endkey，找到该区间内的所有 Region，假设存放在 regionList 中。然后，遍历 regionList，为每一个 Region 调用 Batch.Call，在该接口内，用户定义具体的 RPC 调用逻辑。最后 coprocessorService 将所有 Batch.Call.call()的返回值加入结果集合并返回。

_2019_01_11_3_53_25

coprocessorService 的第三种方法比第二个方法多了一个参数 callback。 coprocessorService 第二个方法内部使用 HBase 自带的缺省 callback，该缺省 callback 将每个 Region 的返回结果都添加到一个 Map 类型的结果集中，并将该集合作为 coprocessorService 方法的返回值。

HBase 提供第三种 coprocessorService 方法允许用户定义 callback 行为，coprocessorService 会为每一个 RPC 返回结果调用该 callback，用户可以在 callback 中执行需要的逻辑，比如执行 sum 累加。用第二种方法的情况下，每个 Region 协处理器 RPC 的返回结果先放入一个列表，所有的 Region 都返回后，用户代码再从该列表中取出每一个结果进行累加;用第三种方法，直接在 callback 中进行累加，省掉了创建结果集合和遍历该集合的开销，效率会更高一些。
因此我们只需要额外定义一个 callback 即可，callback 是一个 Batch.Callback 接口类，用户需要重载其 update 方法。

public S sum(final HTable table, final ColumnInterpreter<R, S, P, Q, T> ci,fin 
al Scan scan)throws Throwable { 
final AggregateRequest requestArg = validateArgAndGetPB(scan, ci, false); 
class SumCallBack implements Batch.Callback { 
S sumVal = null; 
public S getSumResult() { return sumVal; 
} 
@Override public synchronized void update(byte[] region, byte[] row, S result) { 
sumVal = ci.add(sumVal, result); 
}} 
SumCallBack sumCallBack = new SumCallBack(); 
table.coprocessorService(AggregateService.class, scan.getStartRow(), scan.get StopRow(), 
new Batch.Call<AggregateService, S>() {

public S call(AggregateService instance) throws IOException {
ServerRpcController controller = new ServerRpcController(); BlockingRpcCallback<AggregateResponse> rpcCallback =
new BlockingRpcCallback<AggregateResponse>();
//RPC 调用
instance.getSum(controller, requestArg, rpcCallback);
AggregateResponse response = rpcCallback.get(); if (controller.failedOnException()) {
throw controller.getFailedOn(); }
if (response.getFirstPartCount() == 0) { return null;
}
ByteString b = response.getFirstPart(0);
T t = ProtobufUtil.getParsedGenericInstance(ci.getClass(), 4, b); S s = ci.getPromotedValueFromProto(t);
return s;
}
}, sumCallBack);
return sumCallBack.getSumResult();

4. Observer 实现二级索引

Observer 类似于传统数据库中的触发器，当发生某些事件的时候这类协处理器会被 Server 端调用。Observer Coprocessor 是一些散布在 HBase Server 端代码的 hook 钩子，在固定的事件发生时被调用。比如:put 操作之前有钩子函数 prePut，该函数在 pu 操作执行前会被 Region Server 调用;在 put 操作之后则有 postPut 钩子函数。

_2019_01_11_4_59_56