• 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一些最基本的操作,是并没有涉及什么原理之类的, 所以在了解其原理之前这也是必备的知识, 如果本片文章对你有些许帮助那就点个赞吧哈哈