HBase协处理器实战

Posted 2020-08-19

主要内容：

1. HBase协处理器介绍

2. 观察者（Observer）

3. 终端（endpoint）

-------------------------------------------------------------------------------------------------------------------------------------------------------------------

1. HBase协处理器介绍

     系统协处理器可以全局导入region server上的所有数据表，表协处理器即是用户可以指定一张表使用协处理器。Hbase协处理器（Coprocessor）有两种类型：Observer Coprocessors 和Endpoint Coprocessor。

     前者类似触发器，在特定的事件发生时候触发，后者类似存储过程，执行数据计算。观察者协处理器在很多地方可能用到这些，比如：数据安全权限限制，数据外键参考或者一致性，二级索引，主要类型有：RegionObserver，RegionServerObserver，MasterObserver，WalObserver。

2. 观察者（Observer）

      观察者的设计意图是允许用户通过插入代码来重载协处理器框架的upcall方法，而具体的事件触发的callback方法由HBase的核心代码来执行。协处理器框架处理所有的callback调用细节，协处理器自身只需要插入添加或者改变的功能。

     以HBase0.92版本为例，它提供了三种观察者接口：

• RegionObserver：提供客户端的数据操纵事件钩子：Get、Put、Delete、Scan等。
• WALObserver：提供WAL相关操作钩子。
• MasterObserver：提供DDL类型的操作钩子。如创建、删除、修改数据表等。

     这些接口可以同时使用在同一个地方，按照不同优先级顺序执行.用户可以任意基于协处理器实现复杂的HBase功能层。HBase有很多种事件可以触发观察者方法，这些事件与方法从HBase0.92版本起，都会集成在HBase API中。不过这些API可能会由于各种原因有所改动，不同版本的接口改动比较大，具体参考Java Doc，RegionObserver工作原理如下图所示。

图1 RegionObserver工作原理

 

3. 终端（endpoint）

       HBase 提供了客户端 Java 包 org.apache.hadoop.hbase.client.coprocessor。它提供以下三种方法来调用协处理器提供的服务：

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

      Endpoint 协处理器在Region上下文中运行，一个 HBase 表可能有多个Region。因此客户端可以指定调用某一个单个Region上的协处理器，在单个Region上进行处理并返回一定结果；也可以调用一定范围内的若干Region上的协处理器并发执行，并对结果进行汇总处理。针对不同的需要，可以选择以下三种方法。

    （1）调用单个Region上的协处理器RPC

     第一个方法使用API coprocessorService(byte[])，这个函数只调用单个Region上的协处理器。

     该方法采用RowKey指定Region。这是因为HBase的客户端很少会直接操作Region，一般不需要知道 Region 的名字；况且在 HBase 中，Region 名会随时改变，所以用rowkey来指定Region是最合理的方式。使用 rowkey 可以指定唯一的一个Region，如果给定的 rowkey 并不存在，只要在某个Region的rowkey范围内，依然可以用来指定该Region。比如Region1处理[row1, row100]这个区间内的数据，则 rowkey=row1 就由Region1来负责处理，换句话说，我们可以用row1来指定Region1，无论 rowkey 等于”row1”的记录是否存在。

 图2 调用单个Region上的协处理器

      coprocessorService 方法返回类型为 CoprocessorRpcChannel 的对象，该 RPC 通道连接到由 rowkey 指定的 Region 上，通过这个通道，就可以调用该 Region 上部署的协处理器 RPC。我们已经通过 Protobuf 定义了 RPC Service。调用 Service 的 newBlockingStub() 方法，将 CoprocessorRpcChannel 作为输入参数，就可以得到 RPC 调用的 stub 对象，进而调用远端的 RPC。

代码1 获取单个Region的rowcount

 1 long singleRegionCount(String tableName, String rowkey,boolean reCount)
 2 {
 3  long rowcount = 0;
 4  try{
 5  Configuration config = new Configuration();
 6  HConnection conn = HConnectionManager.createConnection(config);
 7  HTableInterface tbl = conn.getTable(tableName);
 8  //获取 Channel
 9  CoprocessorRpcChannel channel = tbl.coprocessorService(rowkey.getBytes());
10 org.ibm.developerworks.getRowCount.ibmDeveloperWorksService.BlockingInterface service =
11 org.ibm.developerworks.getRowCount.ibmDeveloperWorksService.newBlockingStub(channel);
12   //设置 RPC 入口参数
13 org.ibm.developerworks.getRowCount.getRowCountRequest.Builder request = 
14 org.ibm.developerworks.getRowCount.getRowCountRequest.newBuilder();
15  request.setReCount(reCount);
16   //调用 RPC
17  org.ibm.developerworks.getRowCount.getRowCountResponse ret =
18  service.getRowCount(null, request.build());
19  
20  //解析结果
21  rowcount = ret.getRowCount();
22  }
23  catch(Exception e) {e.printStackTrace();}
24  return rowcount;
25  }

     （2）调用多个 Region 上的协处理器 RPC，不使用 callback

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

      这意味着该客户端同时和多个 Region 进行批处理交互。具体方法是，收集每个Region的startkey，然后循环调用第一种coprocessorService方法：用每一个Region的startkey作为入口参数，获得RPC通道，创建 stub对象，进而逐一调用每个Region上的协处理器RPC。这种做法需要写很多的代码，为此HBase提供了两种更加简单的coprocessorService方法来处理多个Region的协处理器调用。先来看第一种方法 coprocessorService(Class, byte[],byte[],Batch.Call)，该方法有 4 个入口参数。第一个参数是实现RPC的Service类，即前文中的ibmDeveloperWorksService类。通过它，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()的返回值加入结果集合并返回。如下图所示：

 

 

图3 调用多个Region上的协处理器——不使用callback

       （3）调用多个 Region 上的协处理器 RPC，使用 callback

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

      这个结果集合的key是Region名字，value是call方法的返回值。采用这种方法，客户端代码需要将RPC执行结果先保存在一个集合中，再进入一个循环，遍历结果集合进一步处理。有些情况下这种使用集合的开销是不必要的。对每个 Region 的返回结果直接进行处理可以省去这些开销。具体过程如下图所示：

图4 调用多个Region上的协处理器——使用callback

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