HBase的Coprocessor是模仿谷歌BigTable的Coprocessor模型实现的。
Coprocessor提供了一种机制可以让开发者直接在RegionServer上运行自定义代码来管理数据。
首先必须要指明使用Coprocessor还是存在一些风险的。Coprocessor是HBase的高级功能，本来是只为HBase系统开发人员准备的。因为Coprocessor的代码直接在RegionServer上运行，并直接接触数据，这样就带来了数据破坏的风险，比如“中间人攻击（Man-in-the-MiddleAttack，简称“MITM攻击”，见百度词条）”以及其他类型的恶意入侵。目前还没有任何机制来屏蔽Coprocessor导致的数据破坏。此外，因为没有资源隔离，一个即使不是恶意设计的但表现不佳的Coprocessor也会严重影响集群的性能和稳定性。
通常我们访问HBase的方式是使用scan或get获取数据，使用Filter过滤掉不需要的部分，最后在获取到的数据上进行业务运算。但是在数据量非常大的时候，比如一个有上亿行及十万个列的数据集，再按常用的方式移动获取数据就会在网络层面遇到瓶颈。客户端也需要有强大的计算能力以及足够的内存来处理这么多的数据。此外，这也会使客户端的代码变得庞大而复杂。
这种场景正是Coprocessor可以发挥作用的地方。我们可以将业务运算代码封装到Coprocessor中并在RegionServer上运行，即在数据实际存储位置执行，最后将运算结果返回到客户端。
如下的一些理论可以帮助我们理解Coprocessor是如何发挥作用的：
触发器和存储过程：一个Observer Coprocessor有些类似于关系型数据库中的触发器，通过它我们可以在一些事件（如Get或是Scan）发生前后执行特定的代码。Endpoint Coprocessor则类似于关系型数据库中的存储过程，因为它允许我们在RegionServer上直接对它存储的数据进行运算，而非是在客户端完成运算。
MapReduce：MapReduce的原则就是将运算移动到数据所处的节点。Coprocessor也是按照相同的原则去工作的。
AOP：如果熟悉AOP的概念的话，可以将Coprocessor的执行过程视为在传递请求的过程中对请求进行了拦截，并执行了一些自定义代码。
Coprocessor类型
Coprocessor可以分为两大类：Observer Coprocessors（观察者）和EndPoint Coprocessor（终端）。
Observer Coprocessors
Observer Coprocessor在一个特定的事件发生前或发生后触发。在事件发生前触发的Coprocessor需要重写以pre作为前缀的方法，比如prePut。在事件发生后触发的Coprocessor使用方法以post作为前缀，比如postPut。
Observer Coprocessor的使用场景如下：
- 安全性：在执行Get或Put操作前，通过preGet或prePut方法检查是否允许该操作；
- 引用完整性约束：HBase并不直接支持关系型数据库中的引用完整性约束概念，即通常所说的外键。但是我们可以使用Coprocessor增强这种约束。比如根据业务需要，我们每次写入user表的同时也要向user_daily_attendance表中插入一条相应的记录，此时我们可以实现一个Coprocessor，在prePut方法中添加相应的代码实现这种业务需求。
- 二级索引：可以使用Coprocessor来维持一个二级索引。这里暂不展开，有时间会单独说明。
根据作用的对象，Observer Coprocessor有如下几种：RegionObserver、RegionServerObserver、MasterObserver和WalObserver。我们可以通过这些Observer来处理其观察的对象的操作，比如可以通过RegionObserver处理Region相关的事件，如Get和Put操作。
Endpoint Coprocessor
Endpoint Coprocessor可以让开发者在数据本地执行运算。一个典型的案例：一个table有几百个Region，需要计算它的运行平均值或者总和。
Observer Coprocessor中代码的执行是相对透明的，而对于Endpoint Coprocessor，则需要显式的调用Table, HTableInterface或者HTable中的CoprocessorService()方法才能使之执行。
从0.96版本开始，HBase开始使用Google的protobuff。这对Endpoint Coprocessor的开发多少有一些影响。Endpoint Coprocessor不应该使用HBase内部成员，尽量只使用公共的API，最理想的情况应该是只依赖接口和数据结构。这样可以使开发的Endpoint Coprocessor更加健壮，不会受到HBase内核演进的干扰。注释为private或evolving的HBase内部API在删除前不必遵守关于deprecate的语义版本规则或相关的一般java规则。而使用protobuff生成的文件不会受到这些注释的影响，因为这些文件是用protoc工具自动生成的。在生成时这些文件时，protoc不知道也不会考虑HBase是如何工作的。
装载和卸载Coprocessor
要使用Coprocessor，就需要先完成对其的装载。这可以静态实现（通过HBase配置文件），也可以动态完成（通过shell或Java API）。
静态装载和卸载Coprocessor
按以下如下步骤可以静态装载自定义的Coprocessor。需要注意的是，如果一个Coprocessor是静态装载的，要卸载它就需要重启HBase。
静态装载步骤如下：
1. 在hbase-site.xml中使用<property>标签定义一个Coprocessor。<property>的子元素<name>的值只能从下面三个中选一个：
- hbase.coprocessor.region.classes 对应 RegionObservers和Endpoints；
- hbase.coprocessor.wal.classes 对应 WALObservers；
- hbase.coprocessor.master.classes 对应MasterObservers。
而<value>标签的内容则是自定义Coprocessor的全限定类名。
下面演示了如何装载一个自定义Coprocessor（这里是在SumEndPoint.java中实现的），需要在每个RegionServer的hbase-site.xml中创建如下的记录：
|
1 2 3 4 |
<property> <name>hbase.coprocessor.region.classes</name> <value>org.myname.hbase.coprocessor.endpoint.SumEndPoint</value> </property> |
如果要装载多个类，类名需要以逗号分隔。HBase会使用默认的类加载器加载配置中的这些类，因此需要将相应的jar文件上传到HBase服务端的类路径下。
使用这种方式加载的Coprocessor将会作用在HBase所有表的全部Region上，因此这样加载的Coprocessor又被称为系统Coprocessor。在Coprocessor列表中第一个Coprocessor的优先级值为Coprocessor.Priority.SYSTEM，其后的每个Coprocessor的值将会按序加一（这意味着优先级会减降低，因为优先级是按整数的自然顺序降序排列的）。
当调用配置的Observer Coprocessor时，HBase将会按照优先级顺序依次调用它们的回调方法。
2. 将代码放到HBase的类路径下。一个简单的方法是将封装好的jar（包括代码和依赖）放到HBase安装路径下的/lib目录中。
3. 重启HBase。
静态卸载的步骤如下：
1. 移除在hbase-site.xml中的配置。
2. 重启HBase。
3. 这一步是可选的，将上传到HBase类路径下的jar包移除。
动态装载Coprocessor
动态装载Coprocessor的一个优势就是不需要重启HBase。不过动态装载的Coprocessor只是针对某个表有效。因此，动态装载的Coprocessor又被称为表级Coprocessor。
此外，动态装载Coprocessor是对表的一次schema级别的调整，因此在动态装载Coprocessor时，目标表需要离线。
动态装载Coprocessor有两种方式：通过HBase Shell和通过Java API。
在下面介绍关于动态装载的部分，假设已经封装好了一个coprocessor.jar的包，里面包含实现代码及所有的依赖，并且已经将这个jar上传到了HDFS中。
通过HBase Shell动态装载和卸载
装载步骤如下
1. 在HBase Shell中disable 掉目标表
|
1 |
hbase> disable 'users' |
2. 使用类似如下的命令加载Coprocessor
|
1 2 3 |
hbase alter 'users', METHOD => 'table_att', 'Coprocessor'=>'hdfs://<namenode>:<port>/ user/<hadoop-user>/coprocessor.jar| org.myname.hbase.Coprocessor.RegionObserverExample|1073741823| arg1=1,arg2=2' |
简单解释下这个命令。这条命令在一个表的table_att中添加了一个新的属性“Coprocessor”。使用的时候Coprocessor会尝试从这个表的table_attr中读取这个属性的信息。这个属性的值用管道符“|”分成了四部分：
- 文件路径：文件路径中需要包含Coprocessor的实现，并且对所有的RegionServer都是可达的。这个路径可以是每个RegionServer的本地磁盘路径，也可以是HDFS上的一个路径。通常建议是将Coprocessor实现存储到HDFS。HBASE-14548允许使用一个路径中包含的所有的jar，或者是在路径中使用通配符来指定某些jar，比如：hdfs://<namenode>:<port>/user/<hadoop-user>/ 或者 hdfs://<namenode>:<port>/user/<hadoop-user>/*.jar。需要注意的是如果是用路径来指定要加载的Coprocessor，这个路径下的所有jar文件都会被加载，不过该路径下的子目录中的jar不会被加载。另外，如果要用路径指定Coprocessor时，就不要再使用通配符了。这些特性在Java API中也得到了支持。
- 类名：Coprocessor的全限定类名。
- 优先级：一个整数。HBase将会使用优先级来决定在同一个位置配置的所有Observer Coprocessor的执行顺序。这个位置可以留白，这样HBase将会分配一个默认的优先级。
- 参数（可选的）：这些值会被传递给要使用的Coprocessor实现。这个项是可选的。
3. enable这个表
|
1 |
hbase(main):003:0> enable 'users' |
4. 检验Coprocessor是否被加载
|
1 |
hbase(main):04:0> describe 'users' |
Coprocessor可以在TABLE_ATTRIBUTES中找到。
加载步骤就是这样。
卸载步骤如下
1. disbale目标表
|
1 |
hbase> disable 'users' |
2. 使用alter命令移除掉Coprocessor
|
1 |
hbase> alter 'users', METHOD => 'table_att_unset', NAME => 'coprocessor$1' |
3. enable目标表
|
1 |
hbase> enable 'users' |
使用Java API动态装载和卸载
装载方式如下
针对不同版本的HBase会有不同的JavaAPI。幸运的是有一个全版本的Java API。下面的代码演示了是如何使用Java API来装载Coprocessor的：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
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); |
0.96及更高版本的HBase还有另一套API。在这套API里，HTableDescriptor的addCoprocessor()方法提供了一种更简单的方式来动态加载Coprocessor：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
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); |
卸载方式如下：
卸载方式就是重新加载表定义信息。重新加载的时候就不需要再使用setValue()方法或者是addCoprocessor()方法设置表的Coprocessor信息了：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
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); |
对于0.96及更高版本的HBase，可以使用HTableDescriptor类的removeCoprocessor()方法。
Coprocessor示例程序
在写示例程序之前，先假设一个场景：我们有一张名为“users”的表，包含personalDet和salaryDet两个列族。这两个列族中分别记录了个人信息和薪资信息的详情。具体如下表：
| personalDet | salaryDet | |||||
| rowkey | name | lastname | dob | gross | net | allowances |
| admin | Admin | Admin | ||||
| cdickens | Charles | Dickens | 02/07/1812 | 10000 | 8000 | 2000 |
| jverne | Jules | Verne | 02/08/1828 | 12000 | 9000 | 3000 |
Observer 实例
现在我们写一个Observer Coprocessor，目标是阻止在对users表进行scan或get时获取admin用户的信息。具体步骤如下：
- 实现RegionObserver接口。
- 重写preGetOp()方法（preGet()方法已经过时）检查客户端有没有查询行键为“admin”的记录。如果有的话直接返回空结果，没有的话，正常处理请求。
- 将代码和依赖封装成一个jar。
- 上传jar到HDFS——HBase需要能访问到这个HDFS。
- 加载Coprocessor。
- 编写测试程序进行测试。
下面是Coprocessor的实现：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
public class RegionObserverExample implements RegionObserver { 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<RegionCoprocessorEnvironment> e, final Get get, final List<Cell> results) throws IOException { 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(); } } } |
重写preGetOp方法将只对Get操作生效，要对scan生效还需要重写preScannerOpen()方法来从scan结果中过滤掉“admin”的信息：
|
1 2 3 4 5 6 7 8 |
@Override public RegionScanner preScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> e, final Scan scan, final RegionScanner s) throws IOException { Filter filter = new RowFilter(CompareOp.NOT_EQUAL, new BinaryComparator(ADMIN)); scan.setFilter(filter); return s; } |
现在代码可以工作了，不过还存在一个问题：如果客户端在scan的时候也使用了Filter，客户端使用的Filter就会被这个FIlter覆盖掉。这不是一个好方法，所以我们可以在查询结果上做手脚，从查询结果中删除掉行键为“admin”的记录：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
@Override public boolean postScannerNext(final ObserverContext<RegionCoprocessorEnvironment> e, final InternalScanner s, final List<Result> results, final int limit, final boolean hasMore) throws IOException { Result result = null; Iterator<Result> iterator = results.iterator(); while (iterator.hasNext()) { result = iterator.next(); if (Bytes.equals(result.getRow(), ROWKEY)) { iterator.remove(); break; } } return hasMore; } |
Endpoint实例
还是对users表进行处理。这次的目标是计算所有员工的薪资的总和。需要编写一个Endpoint Coprocessor，步骤如下：
1. 创建一个“.proto”文件定义服务
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
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); } |
“.proto”是protobuff的对象描述文件，使用前需要先安装protobuff，目前使用的版本应该还是2.5版本。
2. 执行protoc命令，通过“.proto”文件生成Java代码
|
1 2 |
$ mkdir src $ protoc --java_out=src ./sum.proto |
根据文件描述定义将会生成一个名为Sum.java的文件。
3. 编写一个Coprocessor类，实现Coprocessor和CoprocessorService两个接口，并实现接口中定义的方法：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 |
public class SumEndPoint extends SumService implements Coprocessor, CoprocessorService { private RegionCoprocessorEnvironment env; @Override public Service getService() { return this; } @Override public void start(CoprocessorEnvironment env) throws IOException { if (env instanceof RegionCoprocessorEnvironment) { this.env = (RegionCoprocessorEnvironment)env; } else { throw new CoprocessorException("Must be loaded on a table region!"); } } @Override public void stop(CoprocessorEnvironment env) throws IOException { // do mothing } @Override public void getSum(RpcController controller, SumRequest request, RpcCallback done) { Scan scan = new Scan(); 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 = new ArrayList(); 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); } } |
4. 加载Coprocessor
5. 编写客户端代码调用Coprocessor
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 |
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) throws IOException { BlockingRpcCallback rpcCallback = new BlockingRpcCallback(); 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(); } |
部署Coprocessor的原则
更新Coprocessor
更新动态部署的Coprocessor并不是简单地disable表，替换jar，然后重新启用Coprocessor。在JVM中，如果一个类还有引用，我们就无法重新加载它。因为当前的JVM对自定义的Coprocessor还有引用，要完成更新就需要重启JVM，也就是重启RegionSever。
关于日志
Coprocessor框架并没有提供日志相关的API。
注意事项
重复加载Coprocessor
如果我们先静态加载了一个Coprocessor，而后又通过HBase Shell动态加载了一次这个Coprocessor。那么先加载的Coprocessor并不会被覆盖，而是会同时存在两个Coprocessor实例。第二个Coprocessor会有更低的优先级，换句话说，重复加载的第二个Coprocessor实例实际上没有发挥作用。
################
发表评论