[TOC] # 分析 ## 协处理器-coprocessor 协处理器有两种：observer和endpoint Observer允许集群在正常的客户端操作过程中可以有不同的行为表现 Endpoint允许扩展集群的能力，对客户端应用开放新的运算命令   1. 客户端发出put请求 2. 该请求被分派给合适的RegionServer和region 3. coprocessorHost拦截该请求，然后在该表上登记的每个RegionObserver上调用prePut() 4. 如果没有被prePut()拦截，该请求继续送到region，然后进行处理 5. region产生的结果再次被CoprocessorHost拦截，调用postPut() 6. 假如没有postPut()拦截该响应，最终结果被返回给客户端 * Observer的类型 1. RegionObs——这种Observer钩在数据访问和操作阶段，所有标准的数据操作命令都可以被pre-hooks和post-hooks拦截 2. WALObserver——WAL所支持的Observer；可用的钩子是pre-WAL和post-WAL 3. MasterObserver——钩住DDL事件，如表创建或模式修改 ## 二级索引 row key在HBase中是以B+ tree结构化有序存储的，所以scan起来会比较效率。单表以row key存储索引，column value存储id值或其他数据 ，这就是Hbase索引表的结构。 由于HBase本身没有二级索引（Secondary Index）机制，基于索引检索数据只能单纯地依靠RowKey，为了能支持多条件查询，开发者需要将所有可能作为查询条件的字段一一拼接到RowKey中，这是HBase开发中极为常见的做法 比如，现在有一张1亿的用户信息表，建有出生地和年龄两个索引，我想得到一个条件是在杭州出生，年龄为20岁的按用户id正序排列前10个的用户列表。 有一种方案是，系统先扫描出生地为杭州的索引，得到一个用户id结果集，这个集合的规模假设是10万。然后扫描年龄，规模是5万，最后merge这些用户id，去重，排序得到结果。 这明显有问题，如何改良？ 保证出生地和年龄的结果是排过序的，可以减少merge的数据量？但Hbase是按row key排序，value是不能排序的。 变通一下——将用户id冗余到row key里？OK，这是一种解决方案了，这个方案的图示如下：  merge时提取交集就是所需要的列表，顺序是靠索引增加了_id，以字典序保证的。 按索引查询种类建立组合索引。 在方案1的场景中，想象一下，如果单索引数量多达10个会怎么样？10个索引，就要merge 10次，性能可想而知。  解决这个问题需要参考RDBMS的组合索引实现。 比如出生地和年龄需要同时查询，此时如果建立一个出生地和年龄的组合索引，查询时效率会高出merge很多。 当然，这个索引也需要冗余用户id，目的是让结果自然有序。结构图示如下：  这个方案的优点是查询速度非常快，根据查询条件，只需要到一张表中检索即可得到结果list。缺点是如果有多个索引，就要建立多个与查询条件一一对应的组合索引 而索引表的维护如果交给应用客户端，则无疑增加了应用端开发的负担 通过协处理器可以将索引表维护的工作从应用端剥离 * 利用Observer自动维护索引表示例 在社交类应用中，经常需要快速检索各用户的关注列表t_guanzhu，同时，又需要反向检索各种户的粉丝列表t_fensi，为了实现这个需求，最佳实践是建立两张互为反向的表： * 一个表为正向索引关注表 “t_guanzhu”： Rowkey: A-B f1:From f1:To * 另一个表为反向索引粉丝表：“t_fensi”： Rowkey: B—A f1:From f1:To 插入一条关注信息时，为了减轻应用端维护反向索引表的负担，可用Observer协处理器实现：  prePut就是前置put postPut就是后置put # 代码 1. 编写代码 ~~~ package com.study; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver; import org.apache.hadoop.hbase.coprocessor.ObserverContext; import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; import org.apache.hadoop.hbase.regionserver.wal.WALEdit; import java.io.IOException; public class FFCoprocessor extends BaseRegionObserver { @Override public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException { Configuration conf = HBaseConfiguration.create(); Connection conn = ConnectionFactory.createConnection(conf); //二级索引表 Table table = conn.getTable(TableName.valueOf("index_user")); //拦截到原始put对象中的rowkey和列的Value //rowkey //列的value byte[] row = put.getRow(); String rowkey = new String(row); Cell addressCell = put.get("info".getBytes(), "address".getBytes()).get(0); byte[] valueArray = addressCell.getValueArray(); String address = new String(valueArray, addressCell.getValueOffset(), addressCell.getValueLength()); //原本的行键用-连起来的 String[] user_fensi = rowkey.split("-"); //里面的是二级索引表的rowkey,添加到二级索引表 Put putIndex = new Put((user_fensi[1] + "-" + user_fensi[0]).getBytes()); putIndex.addColumn("info".getBytes(), "address".getBytes(),address.getBytes()); table.put(putIndex); table.close(); } } ~~~ 2. 打成jar包“coprocess.jar”上传hdfs ~~~ hadoop fs -put coprocess.jar / ~~~ 3. 创建表 ~~~ create 'user_guanzhu','info' create 'index_user','info' ~~~ 4. 修改schema，注册协处理器 需要先disable一下 ~~~ disable 'user_guanzhu' ~~~ 注意下面的表名,jar包和类名修改为自己的 coprocessor后面是指明协处理器jar包的地址,用|分割,指明对应类的全路径名然后|后跟上标识id ~~~ alter 'user_guanzhu',METHOD => 'table_att','coprocessor'=>'hdfs://master:9000/coprocess.jar|com.study.FFCoprocessor|1001|' ~~~ 启用这个表 ~~~ enable 'user_guanzhu' ~~~ 5. 检查是否注册成功,可以看到注册的协处理器 ~~~ describe 'user_guanzhu' ~~~ 6. 向正向索引表中插入数据进行验证 ~~~ put 'user_guanzhu','zhangsan-liuyifei','info:address','beijing' ~~~ # 删除协处理器 ~~~ 先disable表 disable 'user_guanzhu' 删除,$1表示是第一个协处理器 alter 'user_guanzhu',METHOD=>'table_att_unset',NAME=>'coprocessor$1' 再enable表 enable 'user_guanzhu' ~~~