HBase創建二級索引的一些解決方案

HBase的一級索引就是rowkey，咱們只能經過rowkey進行檢索。若是咱們相對hbase裏面列族的列列進行一些組合查詢，就須要採用HBase的二級索引方案來進行多條件的查詢。
常見的二級索引方案有如下幾種：
1.MapReduce方案
2.ITHBASE方案
3.IHBASE方案
4.Coprocessor方案
5.Solr+hbase方案

MapReduce方案

IndexBuilder：利用MR的方式構建Index
優勢：併發批量構建Index
缺點：不能實時構建Index

舉例：
原表：

row  1      f1:name  zhangsan
row  2      f1:name  lisi
row  3      f1:name  wangwu

索引表：

row     zhangsan    f1:id   1
row     lisi        f1:id   2
row     wangwu      f1:id   3

Demo：

package IndexDouble;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import org.apache.commons.collections.map.HashedMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.MultiTableOutputFormat;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.GenericOptionsParser;


public class IndexBuilder {
    private String rootDir;
    private String zkServer;
    private String port;
    private Configuration conf; 
    private HConnection hConn = null;

    private IndexBuilder(String rootDir,String zkServer,String port) throws IOException{
        this.rootDir = rootDir;
        this.zkServer = zkServer;
        this.port = port;

        conf = HBaseConfiguration.create();
        conf.set("hbase.rootdir", rootDir);
        conf.set("hbase.zookeeper.quorum", zkServer);
        conf.set("hbase.zookeeper.property.clientPort", port);

        hConn = HConnectionManager.createConnection(conf); 
    }

    static class MyMapper extends TableMapper<ImmutableBytesWritable, Put>{

        //記錄了要進行索引的列
        private Map<byte[], ImmutableBytesWritable> indexes = new 
                HashMap<byte[], ImmutableBytesWritable>();

        private String familyName;

        @Override
        protected void map(ImmutableBytesWritable key, Result value,
                Context context) throws IOException, InterruptedException {
            //原始表列
            Set<byte[]> keys = indexes.keySet();

            //索引表的rowkey是原始表的列，索引表的列是原始表的rowkey

            for (byte[] k : keys){

                //得到新建索引表的表名
                ImmutableBytesWritable indexTableName = indexes.get(k);

                //Result存放的是原始表的數據
                //查找到內容             根據列族 和 列 獲得原始表的值
                byte[] val = value.getValue(Bytes.toBytes(familyName), k);

                if (val != null) {
                    //索引表
                    Put put = new Put(val);//索引錶行鍵
                    //列族  列   原始表的行鍵
                    put.add(Bytes.toBytes("f1"),Bytes.toBytes("id"),key.get());
                    context.write(indexTableName, put);
                }
            }

        }

        //真正運行Map以前執行一些處理。
        @Override
        protected void setup(Context context) throws IOException,
                InterruptedException {
            //經過上下文獲得配置
            Configuration conf = context.getConfiguration();

            //得到表名
            String tableName = conf.get("tableName"); 
            //String family = conf.get("familyName");
            //得到列族
            familyName = conf.get("columnFamily");

            //得到列
            String[] qualifiers = conf.getStrings("qualifiers"); 

            for (String qualifier : qualifiers) {
                //創建一個映射，爲每個列建立一個表，表的名字tableName+"-"+qualifier
                //原始表的列    索引表新建表名
                indexes.put(Bytes.toBytes(qualifier), 
                        new ImmutableBytesWritable(Bytes.toBytes(tableName+"-"+qualifier)));
            }

        }   
    }


    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        String rootDir = "hdfs://hadoop1:8020/hbase";
        String zkServer = "hadoop1";
        String port = "2181";

        IndexBuilder conn = new IndexBuilder(rootDir,zkServer,port);

        String[] otherArgs = new GenericOptionsParser(conn.conf, args).getRemainingArgs(); 


        //IndexBuilder: TableName,ColumnFamily,Qualifier
        if(otherArgs.length<3){
            System.exit(-1);
        }
        //表名
        String tableName = otherArgs[0];
        //列族
        String columnFamily = otherArgs[1];

        conn.conf.set("tableName", tableName);
        conn.conf.set("columnFamily", columnFamily);

        //列  可能存在多個列
        String[] qualifiers = new String[otherArgs.length-2];

        for (int i = 0; i < qualifiers.length; i++) {
            qualifiers[i] = otherArgs[i+2];
        }

        //設置列
        conn.conf.setStrings("qualifiers", qualifiers);

        @SuppressWarnings("deprecation")
        Job job = new Job(conn.conf,tableName);

        job.setJarByClass(IndexBuilder.class);

        job.setMapperClass(MyMapper.class);
        job.setNumReduceTasks(0);//因爲不須要執行reduce階段

        job.setInputFormatClass(TableInputFormat.class);
        job.setOutputFormatClass(MultiTableOutputFormat.class);

        Scan scan = new Scan();
        TableMapReduceUtil.initTableMapperJob(tableName,scan, 
                MyMapper.class, ImmutableBytesWritable.class, Put.class, job);

        job.waitForCompletion(true);

    }
}

建立原始表
hbase(main):002:0> create 'studentinfo','f1'
0 row(s) in 0.6520 seconds

=> Hbase::Table - studentinfo


hbase(main):003:0> put 'studentinfo','1','f1:name','zhangsan'
0 row(s) in 0.1640 seconds

hbase(main):004:0> put 'studentinfo','2','f1:name','lisi'
0 row(s) in 0.0240 seconds

hbase(main):005:0> put 'studentinfo','3','f1:name','wangwu'
0 row(s) in 0.0290 seconds

hbase(main):006:0> scan 'studentinfo'
ROW                      COLUMN+CELL
 1                       column=f1:name, timestamp=1436262175823, value=zhangsan
 2                       column=f1:name, timestamp=1436262183922, value=lisi
 3                       column=f1:name, timestamp=1436262189250, value=wangwu
3 row(s) in 0.0530 seconds

建立索引表

hbase(main):007:0> create 'studentinfo-name','f1'
0 row(s) in 0.7740 seconds

=> Hbase::Table - studentinfo-name

執行結果

ITHBASE方案

優勢：ITHBase(Indexed Transactional HBase)是HBase的一個事物型的帶索引的擴展。
缺點：須要重構hbase，幾年沒有更新。
http://github.com/hbase-trx/hbase-transactional-tableindexed

IHBASE方案

**優勢：**IHBase（Indexed HBase）是HBase的一個擴展，用干支持更快的掃描。
缺點：須要重構hbase。
原理：在Memstore滿了之後刷磁盤時，IHBase會進行攔截請求,併爲這個memstore的數據構建索引，索引另外一個CF的方式存儲在表內。scan的時候，IHBase會結合索引列中的標記，來加速scan。
http://github.com/ykulbak/ihbase

Coprocessor方案

HIndex–來自華爲的HBase二級索引
http://github.com/Huawei-Hadoop/hindex

The solution is 100% Java, compatible with Apache HBase 0.94.8, and is open sourced under ASL.

Following capabilities are supported currently.
1.multiple indexes on table,
2.multi column index,
3.index based on part of a column value,
4.equals and range condition scans using index, and
5.bulk loading data to indexed table (Indexing done with bulk load).

Solr+hbase方案

Solr是一個獨立的企業級搜索應用服務器，它對並提供相似幹Web-service的API接口。用戶能夠經過http請求，向搜索引擎服務器提交必定格式的XML文件，生成索引；也能夠經過Http Get操做提出查找請求，並獲得XML格式的返回結果。
Solr是一個高性能，採用Java5開發，基幹Lucene的全文搜索服務器。同時對其進行了擴展，提供了比Lucene更爲豐富的查詢語言，同時實現了可配置、可擴展並對查詢性能進行了優化，而且提供了一個完善的功能節理界面，是一款很是優秀的全文搜索引擎。

HBase不容置疑擁有其優點，但其自己只對rowkey支持毫秒級的快速檢索，對於多字段的組合查詢卻無能爲力。
基於Solr的HBase多條件查詢原理很簡單，將HBase表中涉及條件過濾的字段和rowkey在Solr中創建索引，經過Solr的多條件查詢快速得到符合過濾條件的rowkey值，拿到這些rowkey以後在HBASE中經過指定rowkey進行查詢。