首页 > 代码库 > HBase协处理器实战

HBase协处理器实战

主要内容:

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 方法。

 

 

HBase协处理器实战