码迷,mamicode.com
首页 > 其他好文 > 详细

基于solr实现hbase的二级索引

时间:2014-11-16 22:48:13      阅读:301      评论:0      收藏:0      [点我收藏+]

标签:style   blog   http   io   color   ar   os   使用   sp   

一、目的

    了解hbase的都知道,由于hbase基于行健有序存储,在查询时使用行健十分高效,然后想要实现关系型数据库那样可以随意组合的多条件查询、查询总记录数、分页等就比较麻烦了。想要实现这样的功能,我们可以采用两种方法:

  1. 使用hbase提供的filter,
  2. 自己实现二级索引,通过二级索引查询多符合条件的行健,然后再查询hbase。

    第一种方法不多说了,使用起来很方便,但是局限性也很大,hbase的filter是直接扫记录的,如果数据范围很大,会导致查询速度很慢。所以如果能先使用行健把记录缩小到一个较小范围,那么就比较适合,否则就不适用了。此外该方法不能解决获取总数的为。
    第二种是适用范围就比较广泛了,不过根据实现二级索引的方式解决的问题也不同。这里我们选择solr主要是因为solr可以很轻松实现各种查询(本来就是全文检索引擎)。

二、实现方法

    其实hbase结合solr实现方法还是比较简单的,重点在于一些实现细节上。将hbase记录写入solr的关键就在于hbase提供的Coprocessor,Coprocessor提供了两个实现:endpoint和observerendpoint相当于关系型数据库的存储过程,而observer则相当于触发器。说到这相信大家应该就明白了,我们要利用的就是observer。observer允许我们在记录put前后做一些处理,而我们就是通过postPut将记录同步写入solr(关于Coprocessor具体内容请自行查资料)。

    而写入solr这块就比较简单了,如果是单机就使用ConcurrentUpdateSolrServer,如果是集群就是用CloudSolrServer。不过这里需要注意的是由于CloudSolrServer不像ConcurrentUpdateSolrServer那样内置缓存,默认情况下hbase没写一条数据就会向solr提交一次,这样速度会非常慢(很可能hbase写完很久solr这边还在提交),因此要自己实现一个缓存池,根据hbase的写入速度动态调整,并批量向solr提交。

三、实现代码

    实现方法弄清处置后代码就很容易写了。首先看下Coprocessor的代码:

package com.uboxol.hbase.coprocessor;

import com.uboxol.model.VmMoney;
import com.uboxol.solr.SolrWriter;
import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 * User: guojing
 * Date: 14-10-24
 * Time: 上午11:08
 * To change this template use File | Settings | File Templates.
 */
public class SolrIndexCoprocessorObserver extends BaseRegionObserver {
    private static Logger log = Logger.getLogger(SolrIndexCoprocessorObserver.class);

    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {
        String rowKey = Bytes.toString(put.getRow());
        try {
            Cell cellInnerCode = put.get(Bytes.toBytes("data"), Bytes.toBytes("inner_code")).get(0);
            String innerCode = new String(CellUtil.cloneValue(cellInnerCode));

            Cell cellNodeId = put.get(Bytes.toBytes("data"), Bytes.toBytes("node_id")).get(0);
            String nodeId = new String(CellUtil.cloneValue(cellNodeId));

            Cell cellPayType = put.get(Bytes.toBytes("data"), Bytes.toBytes("pay_type")).get(0);
            String payType = new String(CellUtil.cloneValue(cellPayType));

            Cell cellCts = put.get(Bytes.toBytes("data"), Bytes.toBytes("cts")).get(0);
            String cts = new String(CellUtil.cloneValue(cellCts));

            Cell cellTraSeq = put.get(Bytes.toBytes("data"), Bytes.toBytes("tra_seq")).get(0);
            String traSeq = new String(CellUtil.cloneValue(cellTraSeq));

            cts=cts.replace("-","");
            cts=cts.replace(" ","");
            cts=cts.replace(":","");

            VmMoney vm = new VmMoney();
            vm.setCts(cts);
            vm.setId(new Integer(id));
            vm.setInnerCode(innerCode);
            vm.setNodeId(new Integer(nodeId));
            vm.setPayType(new Integer(payType));
            vm.setRowKey(rowKey);
            vm.setTraSeq(traSeq);

            SolrWriter so = new SolrWriter();
            so.addDocToCache(vm);
        } catch (Exception ex){
            log.info("write "+rowKey+" to solr fail:"+ex.getMessage());
            ex.printStackTrace();
        }
    }

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        String rowKey = Bytes.toString(delete.getRow());
        try {
            SolrWriter so = new SolrWriter();
            so.deleteDoc(rowKey);
        } catch (Exception ex){
            log.info("delete "+rowKey+" from solr fail:"+ex.getMessage());
            ex.printStackTrace();
        }
    }
}

    里边代码很简单,就是在hbase记录写入后和删除后调用SolrWriter进行处理。下边看下SolrWriter的实现:

package com.uboxol.solr;

import com.uboxol.model.VmMoney;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class SolrWriter {
    private static Logger log = Logger.getLogger(SolrWriter.class);

    public static String urlSolr = "";     //solr地址
    private static String defaultCollection = "";  //默认collection
    private static int zkClientTimeOut =0 ;//zk客户端请求超时间
    private static int zkConnectTimeOut =0;//zk客户端连接超时间
    private static CloudSolrServer solrserver = null;

    private static int maxCacheCount = 0;   //缓存大小,当达到该上限时提交
    private static Vector<VmMoney> cache = null;   //缓存
    public  static Lock commitLock =new ReentrantLock();  //在添加缓存或进行提交时加锁

    private static int maxCommitTime = 60; //最大提交时间,s

    static {
        Configuration conf = HBaseConfiguration.create();
        urlSolr = conf.get("hbase.solr.zklist", "192.168.12.1:2181,192.168.12.2:2181,192.168.12.3:2181");
        defaultCollection = conf.get("hbase.solr.collection","collection1");
        zkClientTimeOut = conf.getInt("hbase.solr.zkClientTimeOut", 10000);
        zkConnectTimeOut = conf.getInt("hbase.solr.zkConnectTimeOut", 10000);
        maxCacheCount = conf.getInt("hbase.solr.maxCacheCount", 10000);
        maxCommitTime =  conf.getInt("hbase.solr.maxCommitTime", 60*5);

        log.info("solr init param"+urlSolr+"  "+defaultCollection+"  "+zkClientTimeOut+"  "+zkConnectTimeOut+"  "+maxCacheCount+"  "+maxCommitTime);
        try {
            cache=new Vector<VmMoney>(maxCacheCount);

            solrserver = new CloudSolrServer(urlSolr);
            solrserver.setDefaultCollection(defaultCollection);
            solrserver.setZkClientTimeout(zkClientTimeOut);
            solrserver.setZkConnectTimeout(zkConnectTimeOut);

            //启动定时任务,第一次延迟10执行,之后每隔指定时间执行一次
            Timer timer=new Timer();
            timer.schedule(new CommitTimer(),10*1000,maxCommitTime*1000);
        } catch (Exception ex){
            ex.printStackTrace();
        }

    }

    /**
     * 批量提交
     */
    public void inputDoc(List<VmMoney> vmMoneyList) throws IOException, SolrServerException {
        if (vmMoneyList == null || vmMoneyList.size() == 0) {
            return;
        }
        List<SolrInputDocument> doclist= new ArrayList<SolrInputDocument>(vmMoneyList.size());
        for (VmMoney vm : vmMoneyList) {
            SolrInputDocument doc = new SolrInputDocument();
            doc.addField("id", vm.getId());
            doc.addField("node_id", vm.getNodeId());
            doc.addField("inner_code", vm.getInnerCode());
            doc.addField("pay_type", vm.getPayType());
            doc.addField("rowkey", vm.getRowKey());
            doc.addField("cts", vm.getCts());
            doc.addField("tra_seq", vm.getTraSeq());

            doclist.add(doc);
        }
        solrserver.add(doclist);
    }

    /**
     * 单条提交
     */
    public void inputDoc(VmMoney vmMoney) throws IOException, SolrServerException {
        if (vmMoney == null) {
            return;
        }
        SolrInputDocument doc = new SolrInputDocument();
        doc.addField("id", vmMoney.getId());
        doc.addField("node_id", vmMoney.getNodeId());
        doc.addField("inner_code", vmMoney.getInnerCode());
        doc.addField("pay_type", vmMoney.getPayType());
        doc.addField("rowkey", vmMoney.getRowKey());
        doc.addField("cts", vmMoney.getCts());
        doc.addField("tra_seq", vmMoney.getTraSeq());

        solrserver.add(doc);

    }

    public void deleteDoc(List<String> rowkeys) throws IOException, SolrServerException {
        if (rowkeys == null || rowkeys.size() == 0) {
            return;
        }
        solrserver.deleteById(rowkeys);
    }

    public void deleteDoc(String rowkey) throws IOException, SolrServerException {

        solrserver.deleteById(rowkey);
    }

    /**
     * 添加记录到cache,如果cache达到maxCacheCount,则提交
     */
    public static void addDocToCache(VmMoney vmMoney) {
        commitLock.lock();
        try {
            cache.add(vmMoney);
            log.info("cache commit maxCacheCount:"+maxCacheCount);
            if (cache.size() >= maxCacheCount) {
                log.info("cache commit count:"+cache.size());
                new SolrWriter().inputDoc(cache);
                cache.clear();
            }
        } catch (Exception ex) {
            log.info(ex.getMessage());
        } finally {
            commitLock.unlock();
        }
    }

    /**
     * 提交定时器
     */
    static class CommitTimer extends TimerTask {
        @Override
        public void run() {
            commitLock.lock();
            try {
                if (cache.size() > 0) { //大于0则提交
                    log.info("timer commit count:"+cache.size());
                    new SolrWriter().inputDoc(cache);
                    cache.clear();
                }
            } catch (Exception ex) {
                log.info(ex.getMessage());
            } finally {
                commitLock.unlock();
            }
        }
    }
}

    SolrWriter的重点就在于addDocToCache方法和定时器CommitTimer,addDocToCache会在hbase每次插入数据时将记录插入缓存,并且判断是否达到上限,如果达到则将缓存内所用数据提交到solr,此外CommitTimer 则会每隔一段时间提交一次,以保证缓存内所有数据最终写入solr。
    其他一些辅助代码就不贴了,可自行到github查看:hbase-solr-coprocessor (代码仅作参考,由于业务不同不能直接运行)

四、部署

    这里重点说下hbase的Coprocessor部署的一些问题。部署步骤如下:

  • Coprocessor代码打成jar包,拷贝到所有hbase的region server上,注意jdk一定要1.6,高版本可能会导致无法加载
  • 将hbase的hbase.coprocessor.abortonerror设置成true,待确定Coprocessor运行正常后在改为false。此步骤非必要,但是如果Coprocessor有问题会导致所有region无法启动
  • 于我们实现的Coprocessor是region级的,所以不需要启动,直接通过hbase shell即可加载:
    disable tablename
    
    alter tablename,METHOD => table_att,coprocessor=>jar包路径,本地使用file:///开头,hdfs上的则用hdfs:///开头|1001|参数,多个逗号隔开  
    
    enable tablename

五、总结

    这次hbase+solr的部署前后花了不少时间,其实理论方面都很简单,让人感觉轻而易举,但是实际实现的过程中就会遇到不少问题,就比如写入缓存之类的,如果不去测试,就很容易被忽略。

基于solr实现hbase的二级索引

标签:style   blog   http   io   color   ar   os   使用   sp   

原文地址:http://www.cnblogs.com/good-temper/p/4101948.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!