HBase1.2官方文档——Apache HBase Coprocessors

触发器和存储过程 Triggers and Stored Procedure

一个观察者协处理器类似于一个RDBMS中的触发器，它在一个特定的事件（例如Get或Put）发生之前或之后来执行你的代码。一个终端协处理器类似于RDBMS中的一个存储过程，因为它允许你在当前RegionServer存有的数据上执行自定义运算，而不是在客户端执行运算。

MapReduce

MapReduce以把计算移动到数据所在的位置的原则进行运行。协处理器以相同的原则运行。

面向切片的编程设计 AOP

如果你熟悉 面向切片的编程设计 Aspect Oriented Programming (AOP), 你可以认为协处理器可以您可以在把请求传递到最终目的地之前（或者甚至改变目的地），通过截取请求，然后运行一些自定义代码，然后将请求传递到最终目的地。

1. 你的类应该继承自Coprocessor classes的一种, 如BaseRegionObserver, 或者你的类要实现Coprocessor或者CoprocessorService接口。

2. 加载这个coprocessor, 可以静态加载（通过配置文件）或动态加载（使用HBase Shell命令）。 更多内容参考Loading Coprocessors.

3. 从你的客户端代码中调用coprocessor。HBase对协处理器的处理是透明的。

框架API 在coprocessor包里。

87. Types of Coprocessors

Observer coprocessors 在一个特定的事件发生之前或之后被触发。在一个事件之前发生的观察者使用的方法以一个pre前缀开头, 例如prePut。在一个事件之后发生的观察者覆盖的方法以一个post前缀开头, 例如postPut.

87.1.1. Use Cases for Observer Coprocessors

87.1.2. Types of Observer Coprocessor

终端协处理器允许你在数据存放的位置执行运算。参考Coprocessor Analogy。例如，需要计算一个横跨数百个Region的完整表的运行平均值或总和。

与自定义代码透明地执行的观察者协处理器相比，终端协处理器必须用Table,HTableInterface, 或HTable的CoprocessorService()方法显示调用。

从HBase 0.96开始，终端协处理器用Google Protocol Buffers (protobuf)实现。获取更多关于 protobuf的细节，参考Google’sProtocol Buffer Guide. HBase 0.94中编写的Endpoints Coprocessor 和 0.96 及以后的版本并不兼容。参考HBASE-5448。要把你的HBase集群从0.94或更早的版本升级到0.96及以后的版本，你需要重新实现你的协处理器。

Examplesprovides working examples of endpoint coprocessors.

88. 加载协处理器 Loading Coprocessors

下列步骤用来静态加载你的协处理器。请记住，必须重新启动HBase以卸载一个已被静态加载的coprocessor。

1. 在hbase-site.xml定义一个协处理器，使用 <property> 元素和 <name> <value> 子元素。<name> 应该是下列之一：

   • hbase.coprocessor.region.classesfor RegionObservers and Endpoints.

   • hbase.coprocessor.wal.classesfor WALObservers.

   • hbase.coprocessor.master.classesfor MasterObservers.

     <value> 必须包含你的协处理器实现类的完整类名。

     例如，要加载一个 Coprocessor (实现类是 SumEndPoint.java) 你必须在 RegionServer’s 'hbase-site.xml' 文件中 (一般在 'conf' 目录下)创建以下条目:

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

     如果多个类被指定要加载，类名间要用逗号分隔。框架会试图使用默认类加载器加载所有配置的类。因此，这个jar文件必须位于server端的HBase classpath中。

     通过这种方式加载的Coprocessors将对所有表的所有Region都是激活状态的。这些也被称作 系统Coprocessor。首先列出的Coprocessors 将被赋予Coprocessor.Priority.SYSTEM这样的优先级。列表中每一个后续的coprocessor 将会把它的优先级加一(这是在减小它的优先级，因为优先级是按照整数的自然顺序排列的)。

     当调用注册的观察者时，框架会按照它们的优先级顺序执行它们的回调方法。
     关系被打破。

2. 把你的代码放在HBase的 classpath。做这件事的一个简单的方法是把 jar (包含你的代码和所有依赖)放到HBase安装路径的lib/目录下。

3. 重启 HBase。

1. 在hbase-site.xml中删除协处理器的 <property> 元素，包括子元素。

2. 重启 HBase.

3. 可选的操作是，从HBase的 lib/ 目录下或者从classpath中移除这个协处理器的 JAR 文件。

你可以在不重启HBase的情况下，动态加载协处理器。这看起来比静态加载更好，但是动态加载协处理器的是被加载到一张表上，并且只在加载它们的表上可用。由此，动态加载的协处理器有时被称为表协处理器Table Coprocessor。

另外，动态加载一个协处理器相当于改变表的结构，表必须下线以加载协处理器。

有三种动态加载 Coprocessor 的方法。

88.3.1. Using HBase Shell

88.3.2. Using the Java API (all HBase versions)

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 = newHTableDescriptor(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);

88.3.3. Using the Java API (HBase 0.96+ only)

TableName tableName = TableName.valueOf("users");
String path = "hdfs://<namenode>:<port>/user/<hadoop-user>/coprocessor.jar";
Configuration conf =HBaseConfiguration.create();
HBaseAdmin admin = newHBaseAdmin(conf);
admin.disableTable(tableName);
HTableDescriptor hTableDescriptor = newHTableDescriptor(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);

框架不保证将给定的Coprocessor成功加载。例如，shell命令不保证一个jar文件存在于某个位置，也不验证给定的类是否实际存在于这个Jar文件中。

88.4.1. Using HBase Shell

88.4.2. Using the Java API

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 = newHTableDescriptor(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);

下面的 Observer coprocessor 防止 useradmin的详细信息在对users表的Get和Scan操作中返回。

1. 写一个继承BaseRegionObserver的类。

2. 覆盖preGetOp()方法 (preGet()方法已被弃用了) 用以检查客户端是否已查询了rowkey为admin的内容。如果是，返回一个空的结果。否则，正常处理这个请求。

3. 把你的代码和依赖放到一个JAR 文件中。

4. 把这个JAR放到 HBase可以找到它的HDFS中。

5. 加载这个 Coprocessor.

6. 写一个简单的程序测试它。

以下是上面步骤的实现:

public class RegionObserverExample extendsBaseRegionObserver {

    private static final byte[] ADMIN = Bytes.toBytes("admin");
    private static final byte[] COLUMN_FAMILY = Bytes.toBytes("details");
    private static final byte[] COLUMN = Bytes.toBytes("Admin_det");
    private static final byte[] VALUE = Bytes.toBytes("You can't see Admin details");

    @Override
    public void preGetOp(final ObserverContext e, final Get get, finalList results)
    throwsIOException {

        if(Bytes.equals(get.getRow(),ADMIN)) {
            Cell c =CellUtil.createCell(get.getRow(),COLUMN _FAMILY, COLUMN,
            System.currentTimeMillis(), (byte)4, VALUE);
            results.add(c);
            e.bypass();
        }

        List kvs = newArrayList(results.size());
        for(Cell c : results) {
            kvs.add(KeyValueUtil.ensureKeyValue(c));
        }
        preGet(e, get, kvs);
        results.clear();
        results.addAll(kvs);
    }
}

覆盖preGetOp()将只针对Get操作有效。你还需要覆盖preScannerOpen()方法从scan的结果中过滤admin行。

@Override
public RegionScanner preScannerOpen(final ObserverContext e, finalScan scan,
final RegionScanner s) throwsIOException {

    Filter filter = new RowFilter(CompareOp.NOT_EQUAL, newBinaryComparator(ADMIN));
    scan.setFilter(filter);
    returns;
}

这个方法会有副作用。如果客户端在它的scan中使用了一个过滤器，则客户端的过滤器会被 preScannerOpen()方法中的过滤器替代。这种情况下，覆盖preScannerOpen()的替代方法是，显示地从扫描结果中移除任何admin的结果：

@Override
public boolean postScannerNext(final ObserverContext e, finalInternalScanner s,
final List results, final int limit, final boolean hasMore) throwsIOException {
        Result result = null;
    Iterator iterator =results.iterator();
    while(iterator.hasNext()) {
    result =iterator.next();
        if(Bytes.equals(result.getRow(), ROWKEY)) {
            iterator.remove();
            break;
        }
    }
    returnhasMore;
}

仍然使用users表, 这个例子实现了一个协处理器去计算所有员工收入的总和，使用了一个Endpoint Coprocessor.

1. 创建一个 '.proto' 文件定义你的服务。

   option java_package = "org.myname.hbase.coprocessor.autogenerated";
   option java_outer_classname = "Sum";
   option java_generic_services = true;
   option java_generate_equals_and_hash = true;
   option optimize_for =SPEED;
   message SumRequest {
       required string family = 1;
       required string column = 2;
   }
   
   message SumResponse {
     required int64 sum = 1 [default = 0];
   }
   
   service SumService {
     rpc getSum(SumRequest)
       returns (SumResponse);
   }

2. 执行protoc命令从以上的.proto中生成Java代码。

   $ mkdirsrc
   $ protoc --java_out=src ./sum.proto

   这将生成一个java类Sum.java.

3. 写一个类去继承这个生成的service类，实现Coprocessor和CoprocessorService类, 并且覆盖service()方法。

   如果你从hbase-site.xml加载一个协处理器，然后使用HBase Shell加载一个同名的协处理器，它会被第二次加载。相同的类会存在两次，第二个实例将会有一个更高的ID (因此优先级更低).其结果是，复制的coprocessor实际上被忽略了。

   public class SumEndPoint extends SumService implementsCoprocessor, CoprocessorService {
   
       privateRegionCoprocessorEnvironment env;
   
       @Override
       publicService getService() {
           return this;
       }
   
       @Override
       public void start(CoprocessorEnvironment env) throwsIOException {
           if (env instanceofRegionCoprocessorEnvironment) {
               this.env =(RegionCoprocessorEnvironment)env;
           } else{
               throw new CoprocessorException("Must be loaded on a table region!");
           }
       }
   
       @Override
       public void stop(CoprocessorEnvironment env) throwsIOException {
           //do mothing
   }
   
       @Override
       public voidgetSum(RpcController controller, SumRequest request, RpcCallback done) {
           Scan scan = newScan();
           scan.addFamily(Bytes.toBytes(request.getFamily()));
           scan.addColumn(Bytes.toBytes(request.getFamily()), Bytes.toBytes(request.getColumn()));
           SumResponse response = null;
           InternalScanner scanner = null;
           try{
               scanner =env.getRegion().getScanner(scan);
               List results = newArrayList();
               boolean hasMore = false;
                           long sum = 0L;
                   do{
                           hasMore =scanner.next(results);
                           for(Cell cell : results) {
                               sum = sum +Bytes.toLong(CellUtil.cloneValue(cell));
                        }
                           results.clear();
                   } while(hasMore);
   
                   response =SumResponse.newBuilder().setSum(sum).build();
   
           } catch(IOException ioe) {
               ResponseConverter.setControllerException(controller, ioe);
           } finally{
               if (scanner != null) {
                   try{
                       scanner.close();
                   } catch(IOException ignored) {}
               }
           }
           done.run(response);
       }
   }
   Configuration conf =HBaseConfiguration.create();
   //Use below code for HBase version 1.x.x or above.
   Connection connection =ConnectionFactory.createConnection(conf);
   TableName tableName = TableName.valueOf("users");
   Table table =connection.getTable(tableName);
   
   //Use below code HBase version 0.98.xx or below.
   //HConnection connection = HConnectionManager.createConnection(conf);
   //HTableInterface table = connection.getTable("users");
   
   final SumRequest request = SumRequest.newBuilder().setFamily("salaryDet").setColumn("gross")
                               .build();
   try{
   Map<byte[], Long> results = table.CoprocessorService (SumService.class, null, null,
   new Batch.Call<SumService, Long>() {
       @Override
           public Long call(SumService aggregate) throwsIOException {
   BlockingRpcCallback rpcCallback = newBlockingRpcCallback();
               aggregate.getSum(null, request, rpcCallback);
               SumResponse response =rpcCallback.get();
               return response.hasSum() ? response.getSum() : 0L;
           }
       });
       for(Long sum : results.values()) {
           System.out.println("Sum = " +sum);
       }
   } catch(ServiceException e) {
   e.printStackTrace();
   } catch(Throwable e) {
       e.printStackTrace();
   }

4. 加载这个 Coprocessor.

5. 写一个客户端代码去调用这个 Coprocessor.

Configuration conf =HBaseConfiguration.create();
//Use below code for HBase version 1.x.x or above.
Connection connection =ConnectionFactory.createConnection(conf);
TableName tableName = TableName.valueOf("users");
Table table =connection.getTable(tableName);

//Use below code HBase version 0.98.xx or below.
//HConnection connection = HConnectionManager.createConnection(conf);
//HTableInterface table = connection.getTable("users");
Get get = new Get(Bytes.toBytes("admin"));
Result result =table.get(get);
for(Cell c : result.rawCells()) {
    System.out.println(Bytes.toString(CellUtil.cloneRow(c))
        + "==> " +Bytes.toString(CellUtil.cloneFamily(c))
        + "{" +Bytes.toString(CellUtil.cloneQualifier(c))
        + ":" + Bytes.toLong(CellUtil.cloneValue(c)) + "}");
}
Scan scan = newScan();
ResultScanner scanner =table.getScanner(scan);
for(Result res : scanner) {
    for(Cell c : res.rawCells()) {
        System.out.println(Bytes.toString(CellUtil.cloneRow(c))
        + " ==> " +Bytes.toString(CellUtil.cloneFamily(c))
        + " {" +Bytes.toString(CellUtil.cloneQualifier(c))
        + ":" +Bytes.toLong(CellUtil.cloneValue(c))
        + "}");
    }
}