HBase的协处理器涵盖了两种类似关系型数据库中的应用场景:存储过程和触发器,所以协处理器也分为两种:用来实现存储过程功能的终端程序EndPoint和用来实现触发器功能的观察者Observers

Observer

在hbase2.x的时候,按照之前的继承BaseRegionObserver 是不起作用的,经过我的测试,这个类好像是被移除了,我使用的版本是2.1.1

新的实现可以查看接口Coprocessor来查看,我们来看一下

/**
 * Base interface for the 4 coprocessors - MasterCoprocessor, RegionCoprocessor,
 * RegionServerCoprocessor, and WALCoprocessor.
 * Do NOT implement this interface directly. Unless an implementation implements one (or more) of
 * the above mentioned 4 coprocessors, it'll fail to be loaded by any coprocessor host.
 *
 * Example:
 * Building a coprocessor to observe Master operations.
 * <pre>
 * class MyMasterCoprocessor implements MasterCoprocessor {
 *   &#64;Override
 *   public Optional&lt;MasterObserver> getMasterObserver() {
 *     return new MyMasterObserver();
 *   }
 * }
 * class MyMasterObserver implements MasterObserver {
 *   ....
 * }
 * </pre>
 * Building a Service which can be loaded by both Master and RegionServer
 * <pre>
 * class MyCoprocessorService implements MasterCoprocessor, RegionServerCoprocessor {
 *   &#64;Override
 *   public Optional&lt;Service> getServices() {
 *     return new ...;
 *   }
 * }
 */

这是Coprocessor接口的介绍,非常清楚的介绍了协处理器的用法,那么我就简单的翻译一下下,自己的英语很渣的

有四个接口的父类是coprocessors,他们分别是MasterCoprocessor, RegionCoprocessor, RegionServerCoprocessor, WALCoprocessor.
不要去直接实现coprocessors接口,除非你的实现类实现了上述4个Observer
然后下面的就是一些具体的实现代码

经过之前的介绍,我们知道了Observer的作用就类似一个触发器,当指定操作也就是你实现的方法对应的操作被执行的时候,你定义的Observer逻辑就会去在HBase上去跑,以实现一些控制
当然上面的coprocessors的实现不知四个接口,还有一些其他实现类,比如AccessController,这个提供数据访问和管理的基本授权检查等操作,还有一些自己可以去查看一下

多说无益,直接上需求

 t1表中有f1列族
 向f1中插入name,然后自动计算出name的hash作为此行的password列,不可单独添加password
 可删除name,顺便一起删除password,不可单独删除password

maven

<dependencies>
    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-server</artifactId>
        <version>2.1.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>2.7.7</version>
    </dependency>
</dependencies>

直接上代码

public class MyRegionCoprocessor implements RegionCoprocessor {
    @Override
    public Optional<RegionObserver> getRegionObserver() {
        return Optional.ofNullable(new MyRegionObserver());
    }
    private static class MyRegionObserver implements RegionObserver {
        private static final byte[] FAMILY = Bytes.toBytes("f1");
        private static final byte[] COL_NAME = Bytes.toBytes("name");
        private static final byte[] COL_PASSWORD = Bytes.toBytes("password");
        @Override
        public void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, Durability durability) throws IOException {
            //如果插入的是password列,那么就报错
            List<Cell> passwords = put.get(FAMILY, COL_PASSWORD);
            if (passwords.size() != 0 || !passwords.isEmpty()){
                throw new IllegalArgumentIOException("Password insertion is not allowed !!");
            }
            //如果添加的是name,那么就增加相对应的name的hash的password列
            List<Cell> names = put.get(FAMILY, COL_NAME);
            names.forEach(name -> {
                int hash = Objects.hash(name);
                put.addColumn(FAMILY,COL_PASSWORD,Bytes.toBytes(hash));
            });
        }
        @Override
        public void preDelete(ObserverContext<RegionCoprocessorEnvironment> c, Delete delete, WALEdit edit, Durability durability) throws IOException {
            //阻止删除密码,不可单独删除密码
            List<Cell> passwords = delete.get(FAMILY, COL_PASSWORD);
            if (passwords.size() != 0 || !passwords.isEmpty()){
                throw new IllegalArgumentIOException("Password insertion is not delete !!");
            }
            //到这就代表不是密码了,获取name,然后删除对应的password
            List<Cell> names = delete.get(FAMILY, COL_NAME);
            names.forEach(name -> {
                delete.addColumns(FAMILY,COL_PASSWORD);
            });
        }
    }
}

如果你对上面的Optional类感到陌生,你就记住他是用来判断值是否是空的就可以,如果想进一步了解,你可以参考我的专辑Java8学习,希望能帮助到你
上面的代码逻辑还是比较清晰的,在外部类的重写的方法getRegionObserver中返回了我们自己实现的Observer类,当然这个方法是不用必须重写的,当我们需要实现自己的Observer的时候必须重写getRegionObserver,而当我们实现自己的EndPoint的时候,就需要重写getEndpointObserver方法了,还有一个在RegionCoprocessor中的方法是getBulkLoadObserver,从名字我们可以看到是批加载用的
当我们实现好了自己的Observer的时候,我们就需要部署到HBase集群中去看看效果,在网上我并没有找到类似本地测试Observer的办法,这可真是麻烦.

开始部署,当然打成jar,我就不说了,打完后,部署到HBase的方式有两种,一种是通过配置文件的方式去部署,一种是通过动态的方式去部署,前者称为静态部署,后面即动态部署, 静态部署 我将简单介绍一下,如下,而自己使用的是动态部署的方法

在hbase-site.xml中配置静态部署
  <property>
      <name>hbase.coprocessor.region.classes</name>   #这是配置region的协处理器,对应的就是:hbase.coprocessor.master.classes
      <value>你的全类名</value>   #如果想同时配置多个协处理器,可以用逗号分隔多个协处理器的类名
  </property>
相关的配置
  hbase.coprocessor.enabled:是否启用协处理器机制,默认true,即开启
  hbase.coprocessor.user.enabled:即是否允许用户动态配置,如果为false,就只能在xml中配置了
  hbase.coprocessor.wal.classes:即监控wal的
  hbase.coprocessor.abortonerror:即如果个别协处理器启动失败整个hbase是否启动,默认是如果有的协处理器启动失败,那么hbase就不起了
需要注意的是
  1.配置顺序决定了执行顺序
  2. 所有协处理器是以系统级优先级加载的 (优先级一会儿会提到)
  3. 重启集群起作用

动态配置
- 首先你如果使用我上面的代码测试,那么请create 't1','f1'
- 然后将你的jar上传到集群,然后上传到hdfs,如果你使用我的代码测试,请直接将jar上传到hdfs:///hbase.jar
- 然后disable 't1'
- 然后使用特定命令在表上配置协处理器:alter 't1','coprocessor'=>'hdfs:///hbase.jar|qidai.MyRegionCoprocessor|1001|''
- 再然后enabled 't1'
- 然后desc 't1',你就会看到
```
hbase(main):086:0> desc 't1'
Table t1 is ENABLED
#注意这里代表已经配置上去了,如果没有配置上的话,就是 : 紧接着COLUMN FAMILIES DESCRIPTION 字眼,并没有下面的说明
t1, {TABLE_ATTRIBUTES => {coprocessor$1 => 'hdfs://hbase.jar|qidai.MyRegionCoprocessor|1001|'}        
COLUMN FAMILIES DESCRIPTION                                                       
```
- 到了这一步其实就可以了,然后在t1表中尽情测试需求就好了,我测试的时候就很简单的测试了一下,如果你使用我的代码测试,如果有问题,请及时指正哈!!!如果你发现你的动态配置貌似没起作用,那么请重启一下对应的RegionServer,我出现了这个问题,估计是我的笔记本内存上限了,没反应过来??????

到这就该 说说优先级 了,如上如果配置了处理器,多出来的那一行我们分析一下

alter 't1','coprocessor'=>'hdfs:///hbase.jar|qidai.MyRegionCoprocessor|1001|'
alter '表名','coprocessor'=>'hdfs地址|你的全类名|优先级|传入给你实现类的参数'

这里的优先级是0是最高的,以整数来表示的,越小值越先被执行

好了到这之后,大概的一些问题已经解决了,那么怎么将Observer给卸载呢 ?
- 静态部署的话,那么删除对应的xml内容,然后重启HBase即可
- 动态部署的话,我们可以先disable,然后使用alter 't1', METHOD => 'table_att_unset', NAME => 'coprocessor$1'来进行卸载,在最后面的coprocessor$1需要自己改一下,看看自己的处理器名是啥就写啥,然后在enable就行了
剩余的EndPoint的使用这篇暂时不介绍, 不过迟早会补上的 , 因为现在只对Avro了解一些, 而Protobuf 不太清楚, 恩..今天这篇涉及的只是Observer一些最基本的操作,是并没有涉及什么原理之类的, 所以在了解其原理之前这也是必备的知识, 如果本片文章对你有些许帮助那就点个赞吧哈哈

Hbase-Observer

Observer

热门文章

最新文章

相关电子书