Hbase 协处理器

• 权限校验 ：在执行 Get 或 Put 操作之前，您可以使用 preGet 或 prePut 方法检查权限；
• 完整性约束 ： HBase 不支持关系型数据库中的外键功能，可以通过触发器在插入或者删除数据的时候，对关联的数据进行检查；
• 二级索引 ： 可以使用协处理器来维护二级索引。

• RegionObserver : 允许您观察 Region 上的事件，例如 Get 和 Put 操作。
• RegionServerObserver : 允许您观察与 RegionServer 操作相关的事件，例如启动，停止或执行合并，提交或回滚。
• MasterObserver : 允许您观察与 HBase Master 相关的事件，例如表创建，删除或 schema 修改。
• WalObserver : 允许您观察与预写日志（WAL）相关的事件。

• 客户端发出 put 请求
• 该请求被分派给合适的 RegionServer 和 region
• coprocessorHost 拦截该请求，然后在该表的每个 RegionObserver 上调用 prePut()
• 如果没有被 prePut() 拦截，该请求继续送到 region，然后进行处理
• region 产生的结果再次被 CoprocessorHost 拦截，调用 postPut()
• 假如没有 postPut() 拦截该响应，最终结果被返回给客户端

If you are familiar with Aspect Oriented Programming (AOP), you can think of a coprocessor as applying advice by intercepting a request and then running some custom code,before passing the request on to its final destination (or even changing the destination).

如果您熟悉面向切面编程（AOP），您可以将协处理器视为通过拦截请求然后运行一些自定义代码来使用 Advice，然后将请求传递到其最终目标（或者更改目标）。

• 静态加载的协处理器称之为 System Coprocessor （系统级协处理器）,作用范围是整个 HBase 上的所有表，需要重启 HBase 服务；
• 动态加载的协处理器称之为 Table Coprocessor （表处理器），作用于指定的表，不需要重启 HBase 服务。

1. 在 hbase-site.xml 定义需要加载的协处理器。

   <property>
       <name>hbase.coprocessor.region.classes</name>
       <value>org.myname.hbase.coprocessor.endpoint.SumEndPoint</value>
   </property>
   

2. <name> 标签的值必须是下面其中之一：
   1. RegionObservers 和 Endpoints 协处理器：hbase.coprocessor.region.classes
   2. WALObservers 协处理器： hbase.coprocessor.wal.classes
   3. MasterObservers 协处理器：hbase.coprocessor.master.classes <value> 必须是协处理器实现类的全限定类名。如果为加载指定了多个类，则类名必须以逗号分隔。
3. 将 jar(包含代码和所有依赖项) 放入 HBase 安装目录中的 lib 目录下；
4. 重启 HBase。

1. 从 hbase-site.xml 中删除配置的协处理器的元素及其子元素；
2. 从类路径或 HBase 的 lib 目录中删除协处理器的 JAR 文件（可选）；
3. 重启 HBase。

以下示例基于两个前提：

1. coprocessor.jar 包含协处理器实现及其所有依赖项。
2. JAR 包存放在 HDFS 上的路径为：hdfs：// ： / user / /coprocessor.jar

1. 使用 HBase Shell 禁用表

   hbase > disable 'tableName'
   

2. 使用如下命令加载协处理器

   hbase > alter 'tableName', METHOD => 'table_att', 'Coprocessor'=>'hdfs://<namenode>:<port>/
   user/<hadoop-user>/coprocessor.jar| org.myname.hbase.Coprocessor.RegionObserverExample|1073741823|
   arg1=1,arg2=2'
   

   1. Coprocessor` 包含由管道（|）字符分隔的四个参数，按顺序解释如下：
      1. JAR 包路径 ：通常为 JAR 包在 HDFS 上的路径。关于路径以下两点需要注意：
      2. 允许使用通配符，例如：hdfs://<namenode>:<port>/user/<hadoop-user>/*.jar 来添加指定的 JAR 包；
      3. 可以使指定目录，例如：hdfs://<namenode>:<port>/user/<hadoop-user>/ ，这会添加目录中的所有 JAR 包，但不会搜索子目录中的 JAR 包。
      4. 类名 ：协处理器的完整类名。
      5. 优先级 ：协处理器的优先级，遵循数字的自然序，即值越小优先级越高。可以为空，在这种情况下，将分配默认优先级值。
      6. 可选参数 ：传递的协处理器的可选参数。
3. 启用表

   hbase > enable 'tableName'
   

4. 验证协处理器是否已加载

   hbase > describe 'tableName'
   

1. 禁用表

   hbase> disable 'tableName'
   

2. 移除表协处理器

   hbase> alter 'tableName', METHOD => 'table_att_unset', NAME => 'coprocessor$1'
   

3. 启用表

   hbase> enable 'tableName'
   

TableName tableName = TableName.valueOf("users");
String path = "hdfs://<namenode>:<port>/user/<hadoop-user>/coprocessor.jar";
Configuration conf = HBaseConfiguration.create();
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
admin.disableTable(tableName);
HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
HColumnDescriptor columnFamily1 = new HColumnDescriptor("personalDet");
columnFamily1.setMaxVersions(3);
hTableDescriptor.addFamily(columnFamily1);
HColumnDescriptor columnFamily2 = new HColumnDescriptor("salaryDet");
columnFamily2.setMaxVersions(3);
hTableDescriptor.addFamily(columnFamily2);
hTableDescriptor.setValue("COPROCESSOR$1", path + "|"
+ RegionObserverExample.class.getCanonicalName() + "|"
+ Coprocessor.PRIORITY_USER);
admin.modifyTable(tableName, hTableDescriptor);
admin.enableTable(tableName);

TableName tableName = TableName.valueOf("users");
Path path = new Path("hdfs://<namenode>:<port>/user/<hadoop-user>/coprocessor.jar");
Configuration conf = HBaseConfiguration.create();
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
admin.disableTable(tableName);
HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
HColumnDescriptor columnFamily1 = new HColumnDescriptor("personalDet");
columnFamily1.setMaxVersions(3);
hTableDescriptor.addFamily(columnFamily1);
HColumnDescriptor columnFamily2 = new HColumnDescriptor("salaryDet");
columnFamily2.setMaxVersions(3);
hTableDescriptor.addFamily(columnFamily2);
hTableDescriptor.addCoprocessor(RegionObserverExample.class.getCanonicalName(), path,
Coprocessor.PRIORITY_USER, null);
admin.modifyTable(tableName, hTableDescriptor);
admin.enableTable(tableName);

TableName tableName = TableName.valueOf("users");
String path = "hdfs://<namenode>:<port>/user/<hadoop-user>/coprocessor.jar";
Configuration conf = HBaseConfiguration.create();
Connection connection = ConnectionFactory.createConnection(conf);
Admin admin = connection.getAdmin();
admin.disableTable(tableName);
HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
HColumnDescriptor columnFamily1 = new HColumnDescriptor("personalDet");
columnFamily1.setMaxVersions(3);
hTableDescriptor.addFamily(columnFamily1);
HColumnDescriptor columnFamily2 = new HColumnDescriptor("salaryDet");
columnFamily2.setMaxVersions(3);
hTableDescriptor.addFamily(columnFamily2);
admin.modifyTable(tableName, hTableDescriptor);
admin.enableTable(tableName);

# redis append 命令示例
redis>  EXISTS mykey
(integer) 0
redis>  APPEND mykey "Hello"
(integer) 5
redis>  APPEND mykey " World"
(integer) 11
redis>  GET mykey 
"Hello World"

# 创建一张杂志表 有文章和图片两个列族
hbase >  create 'magazine','article','picture'

完整代码可见本仓库：hbase-observer-coprocessor

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-common</artifactId>
    <version>1.2.0</version>
</dependency>
<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-server</artifactId>
    <version>1.2.0</version>
</dependency>

public class AppendRegionObserver extends BaseRegionObserver {
    private byte[] columnFamily = Bytes.toBytes("article");
    private byte[] qualifier = Bytes.toBytes("content");
    @Override
    public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit,
                       Durability durability) throws IOException {
        if (put.has(columnFamily, qualifier)) {
            // 遍历查询结果，获取指定列的原值
            Result rs = e.getEnvironment().getRegion().get(new Get(put.getRow()));
            String oldValue = "";
            for (Cell cell : rs.rawCells())
                if (CellUtil.matchingColumn(cell, columnFamily, qualifier)) {
                    oldValue = Bytes.toString(CellUtil.cloneValue(cell));
                }
            // 获取指定列新插入的值
            List<Cell> cells = put.get(columnFamily, qualifier);
            String newValue = "";
            for (Cell cell : cells) {
                if (CellUtil.matchingColumn(cell, columnFamily, qualifier)) {
                    newValue = Bytes.toString(CellUtil.cloneValue(cell));
                }
            }
            // Append 操作
            put.addColumn(columnFamily, qualifier, Bytes.toBytes(oldValue + newValue));
        }
    }
}

# mvn clean package

# 上传项目到HDFS上的hbase目录
hadoop fs -put /usr/app/hbase-observer-coprocessor-1.0-SNAPSHOT.jar /hbase
# 查看上传是否成功
hadoop fs -ls /hbase

1. 加载协处理器前需要先禁用表

   hbase >  disable 'magazine'
   

2. 加载协处理器

   hbase >   alter 'magazine', METHOD => 'table_att', 'Coprocessor'=>'hdfs://hadoop001:8020/hbase/hbase-observer-coprocessor-1.0-SNAPSHOT.jar|com.heibaiying.AppendRegionObserver|1001|'
   

3. 启用表

   hbase >  enable 'magazine'
   

4. 查看协处理器是否加载成功

   hbase >  desc 'magazine'
   

hbase > put 'magazine', 'rowkey1','article:content','Hello'
hbase > get 'magazine','rowkey1','article:content'
hbase > put 'magazine', 'rowkey1','article:content','World'
hbase > get 'magazine','rowkey1','article:content'

hbase > put 'magazine', 'rowkey1','article:author','zhangsan'
hbase > get 'magazine','rowkey1','article:author'
hbase > put 'magazine', 'rowkey1','article:author','lisi'
hbase > get 'magazine','rowkey1','article:author'

1. 卸载协处理器前需要先禁用表

   hbase >  disable 'magazine'
   

2. 卸载协处理器

   hbase > alter 'magazine', METHOD => 'table_att_unset', NAME => 'coprocessor$1'
   

3. 启用表

   hbase >  enable 'magazine'
   

4. 查看协处理器是否卸载成功

   hbase >  desc 'magazine'
   

hbase > get 'magazine','rowkey1','article:content'
hbase > put 'magazine', 'rowkey1','article:content','Hello'
hbase > get 'magazine','rowkey1','article:content'

1. Apache HBase Coprocessors
2. Apache HBase Coprocessor Introduction
3. HBase 高階知識