Hbase Coprocessor 协处理器与 JavaAPI

协处理器概念

一、协处理器有两种： observer 和 endpoint

1、observer协处理器

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

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

RegionObserver：提供客户端的数据操纵事件钩子： Get、 Put、 Delete、 Scan 等
WALObserver：提供 WAL 相关操作钩子。
MasterObserver：提供 DDL-类型的操作钩子。如创建、删除、修改数据表等。
到 0.96 版本又新增一个 RegionServerObserver下图是以 RegionObserver 为例子讲解 Observer 这种协处理器的原理：

客户端发起get请求
该请求被分派给合适的RegionServer和Region
coprocessorHost拦截该请求，然后在该表上登记的每个RegionObserer上调用preGet()
如果没有被preGet拦截，该请求继续送到Region，然后进行处理
Region产生的结果再次被coprocessorHost拦截，调用posGet()处理
加入没有postGet()拦截该响应，最终结果被返回给客户端

2、endpoint 协处理器

Endpoint 协处理器类似传统数据库中的存储过程，客户端可以调用这些 Endpoint 协处理器执行一段 Server 端代码，并将 Server 端代码的结果返回给客户端进一步处理，最常见的用法就是进行聚集操作
如果没有协处理器，当用户需要找出一张表中的最大数据，即max 聚合操作，就必须进行全表扫描，在客户端代码内遍历扫描结果，并执行求最大值的操作。这样的方法无法利用底层集群的并发能力，而将所有计算都集中到 Client 端统一执行，势必效率低下。
利用 Coprocessor，用户可以将求最大值的代码部署到 HBase Server 端，HBase 将利用底层 cluster 的多个节点并发执行求最大值的操作。即在每个 Region 范围内执行求最大值的代码，将每个 Region 的最大值在 Region Server 端计算出，仅仅将该 max 值返回给客户端。在客户端进一步将多个 Region 的最大值进一步处理而找到其中的最大值。这样整体的执行效率就会提高很多

协处理器 Java API

一、pom 配置

    <!-- https://mvnrepository.com/artifact/org.apache.hbase/hbase-client --><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-client</artifactId><version>2.2.4</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.hbase/hbase-server --><dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-server</artifactId><version>2.2.4</version></dependency>

二、在自定义包里创建类

在此文件包名为 com.niitchina.hbasedemo.coprocessor,类名为MyRegionObserver，此路径与名字与后面配置有密切联系，若想偷懒可以完全跟随。

三、写入代码

package com.niitchina.hbasedemo.coprocessor;import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.coprocessor.*;
import org.apache.hadoop.hbase.regionserver.FlushLifeCycleTracker;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.regionserver.Store;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.wal.WALEdit;import java.io.FileWriter;
import java.io.IOException;
import java.util.List;
import java.util.Optional;public class MyRegionObserver implements RegionObserver,RegionCoprocessor {@Overridepublic Optional<RegionObserver> getRegionObserver() {return Optional.of(this);}private static void outInfo(String str){try {FileWriter fw = new FileWriter("/training/hbase-2.2.4/coprocessor.txt",true);fw.write(str + "\r\n");fw.close();} catch (Exception e) {e.printStackTrace();}}@Overridepublic void start(CoprocessorEnvironment env) throws IOException {RegionCoprocessor.super.start(env);outInfo("MyRegionObserver.start()");}@Overridepublic void stop(CoprocessorEnvironment env) throws IOException {}@Overridepublic void preGetOp(ObserverContext<RegionCoprocessorEnvironment> e, Get get, List<Cell> results) throws IOException {RegionObserver.super.preGetOp(e, get, results);String rowkey = Bytes.toString(get.getRow());// custom code here , this code will run before the get operationoutInfo("MyRegionObserver.preGetOp() : Before get operation rowkey = " + rowkey);}public void postGetOp(ObserverContext<RegionCoprocessorEnvironment> e, Get get, List<Cell> results) throws IOException {RegionObserver.super.postGetOp(e, get, results);String rowkey = Bytes.toString(get.getRow());//custom codeoutInfo("MyRegionObserver.postGetOp() : After Get Operation rowkey = " + rowkey);}@Overridepublic void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, Durability durability) throws IOException {RegionObserver.super.prePut(c, put, edit, durability);String rowkey = Bytes.toString(put.getRow());// logicoutInfo("MyRegionObserver.prePut() : rowkey = " + rowkey);}@Overridepublic void postPut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, Durability durability) throws IOException {RegionObserver.super.postPut(c, put, edit, durability);String rowkey = Bytes.toString(put.getRow());// custom codeoutInfo("MyRegionObserver.postPut() : rowkey = " + rowkey);}@Overridepublic void preDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {RegionObserver.super.preDelete(e, delete, edit, durability);String rowkey = Bytes.toString(delete.getRow());outInfo("MyRegionObserver.preDelete() : rowkey = " + rowkey);}@Overridepublic void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {RegionObserver.super.postDelete(e, delete, edit, durability);String rowkey = Bytes.toString(delete.getRow());// custom codeoutInfo("MyRegionObserver.postDelete() : rowkey = " + rowkey);}
}

四、项目打jar包

五、Hbase 文件配置

1、进入Hbase安装路径下conf文件夹对hbase-site.xml进行配置，com.niitchina.hbasedemo.coprocessor为包名，MyRegionObserver为类名，若上述步骤名称不同则这里需要更换具体内容。

<property>
<name>hbase.coprocessor.region.classes</name>
<value>com.niitchina.hbasedemo.coprocessor.MyRegionObserver</value>
</property>

2.、重启Hbase

六、运行插入、删除、查询get可以与协处理器连接

日志文件将会保存在自定义路径下，我们代码里设置的 /training/hbase-2.2.4,可以看到日志文件出来了

vi coprocessor.txt 可以看看里面日志，运行成功。