首页 > 其他分享 >(HBase+Lucene)

(HBase+Lucene)

时间:2023-03-28 15:02:31浏览次数:32  
标签:lucene String org Lucene static new import HBase


1、核心工具类



package junit;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.Term;
import org.apache.lucene.queryParser.MultiFieldQueryParser;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.Query;
import org.apache.lucene.search.ScoreDoc;
import org.apache.lucene.search.TopDocs;
import org.apache.lucene.search.highlight.Formatter;
import org.apache.lucene.search.highlight.Fragmenter;
import org.apache.lucene.search.highlight.Highlighter;
import org.apache.lucene.search.highlight.QueryScorer;
import org.apache.lucene.search.highlight.SimpleFragmenter;
import org.apache.lucene.search.highlight.SimpleHTMLFormatter;
import org.apache.lucene.store.FSDirectory;
import org.apache.lucene.util.Version;
import org.junit.Test;
import org.wltea.analyzer.lucene.IKAnalyzer;

import com.ljq.entity.Person;
import com.ljq.utils.HBaseAPI;
import com.ljq.utils.LuceneUtil;
import com.ljq.utils.XMLPropertyConfig;

/**   
 *  模拟功能:假如从互联网上采集数据,采集到的数据实时存储到HBase数据库上,<br/>
 *  现需把这些数据从HBase读取出来,并在本地创建lucene索引,<br/>方便前端通过索引进行查询,把查询取得的结果在界面上展现出来。<br/><br/>
 *  
 *  要求该功能要7*24不间断执行,实时性要求非常高,故采用Strom实时计算框架为载体,实现上面的模拟功能。
 *
 * @author 林计钦
 * @version 1.0 2013-6-5 下午05:36:09   
 */
public class IndexCreationTest {
    private static StringBuffer ids=new StringBuffer();
    private static IndexWriter indexWriter = LuceneUtil.getIndexWriter();
    // 计算时间评估参数
    private static long _initStartTime = 0; // 获取CAS consumer被实例化时的系统时间
    private static long mTimeForCommit = 60; // 经过多久时间提交
    // 计数器
    private static int mIndexCommitBatchNum = 100; // 批量更新索引
    private static int mDocCount = 0;
    /**
     * 从HBase实时读取数据,并在本地创建lucene索引 
     */
    @Test
    public void createIndexFromHBase(){
        readTable("1370422328578");
    }
    
    /**
     * 查询索引
     * @throws IOException 
     */
    @Test
    public void searchIndex() throws IOException{
        for(int i=0;i<500;i++){
            search(new String[]{"id", "name", "age", "rowKey"}, i+"");
        }
//        search(new String[]{"id", "name", "age", "rowKey"}, "0");
//        search(new String[]{"id", "name", "age", "rowKey"}, "998");
        
        //数据写入txt文件
        BufferedWriter writer = new BufferedWriter(new FileWriter(new File("E:\\123.txt")));
        writer.write(ids.toString());
        writer.close();
    }
    
    /**
     * 从HBase实时读取数据,并在本地创建lucene索引
     * 
     * @param startRow 指定开始的行(时间戳)
     */
    public static void readTable(String startRow) {
        int i = 0;
        try {
            while (true) {
                // [{"timestamp":"1370422360734","id":"950","name":"lin950","age":"950","row":"1370422507578"}]
                System.out.println("startRow=" + startRow);
                List<Map<String, String>> datas = HBaseAPI.scan("tb_stu", null, String
                        .valueOf(startRow), 100);
                i += datas.size();
                System.out.println(i);
                if (datas != null && datas.size() > 0) {
                    for (Map<String, String> data : datas) {
                        String row = data.get(HBaseAPI.HBASE_ROW);
                        startRow = row;
                        // System.out.println(String.format("id:%s, name:%s,
                        // age:%s, rowKey:%s.",
                        // data.get("id"), data.get("name"), data.get("age"),
                        // row));
                        createIndex(data.get("id"), data.get("name"), data.get("age"), row);
                    }
                }
                
                if(isCommitTime()){
                    System.out.println("indexCommit");
                    try {
                        indexWriter.commit(); // 批量提交
                        indexWriter.forceMerge(1); // forceMerge代替optimize
                        System.out.println("indexWriter.commit();");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }  
                    System.out.println("indexCommit fin");
                }
            }
        }catch(Exception e){
            e.printStackTrace();
        }
    }
    
    private static void createIndex(String id, String name, String age, String rowKey){
        if (StringUtils.isBlank(id) || StringUtils.isBlank(name) || StringUtils.isBlank(age)
                || StringUtils.isBlank(rowKey)) {
            System.out.println(String.format("id:%s, name:%s, age:%s, rowKey:%s.", 
                id, name, age, rowKey));
            return;
        }
        try {
            Document doc = new Document();
            doc.add(new Field("id", id, Field.Store.YES, Field.Index.NOT_ANALYZED));
            doc.add(new Field("name", name, Field.Store.YES, Field.Index.ANALYZED));
            doc.add(new Field("age", age, Field.Store.YES, Field.Index.NOT_ANALYZED));
            doc.add(new Field("rowKey", rowKey, Field.Store.YES, Field.Index.NOT_ANALYZED));
            
            //更新索引
            if(LuceneUtil.existsIndex()){
                System.out.println("---update index---");
                indexWriter.updateDocument(new Term("id", id), doc);
            }else { //第一次创建索引
                System.out.println("---create index---");
                indexWriter.addDocument(doc);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    /**
     * 判断是否可以提交索引
     * @return
     */
    private static synchronized boolean isCommitTime(){
        //每隔1分钟提交一次
        if((System.currentTimeMillis() - _initStartTime) >= (mTimeForCommit*1000)){
            _initStartTime = System.currentTimeMillis();
            return true;
        }
        //累加到100条提交一次
        else if(mDocCount % mIndexCommitBatchNum == 0){
            _initStartTime = System.currentTimeMillis();
            return true ;
        }
        else
           return false;
    }
    
    /**
     * 搜索、高亮显示
     * 
     * @param fields
     * @param keyword
     */
    private void search(String[] fields, String keyword) {
        IndexSearcher indexSearcher = null;

         
        try {
            // 创建索引搜索器,且只读
            IndexReader indexReader = IndexReader.open(FSDirectory.open(new File(XMLPropertyConfig.getConfigXML()
                    .getString("index_path"))), true);
            
            indexSearcher = new IndexSearcher(indexReader);

            MultiFieldQueryParser queryParser = new MultiFieldQueryParser(Version.LUCENE_35,
                    fields, new IKAnalyzer());
            Query query = queryParser.parse(keyword);

            // 返回前number条记录
            TopDocs topDocs = indexSearcher.search(query, 1000);
            // 信息展示
            int totalCount = topDocs.totalHits;
            //System.out.println("共检索出 " + totalCount + " 条记录");

            // 高亮显示
            /*
             * 创建高亮器,使搜索的结果高亮显示 SimpleHTMLFormatter:用来控制你要加亮的关键字的高亮方式 此类有2个构造方法
             * :SimpleHTMLFormatter()默认的构造方法.加亮方式:<B>关键字</B>
             * :SimpleHTMLFormatter(String preTag, String
             * postTag).加亮方式:preTag关键字postTag
             */
            Formatter formatter = new SimpleHTMLFormatter("<font color='red'>", "</font>");
            /*
             * QueryScorer QueryScorer
             * 是内置的计分器。计分器的工作首先是将片段排序。QueryScorer使用的项是从用户输入的查询中得到的;
             * 它会从原始输入的单词、词组和布尔查询中提取项,并且基于相应的加权因子(boost factor)给它们加权。
             * 为了便于QueryScoere使用,还必须对查询的原始形式进行重写。 比如,带通配符查询、模糊查询、前缀查询以及范围查询
             * 等,都被重写为BoolenaQuery中所使用的项。
             * 在将Query实例传递到QueryScorer之前,可以调用Query.rewrite
             * (IndexReader)方法来重写Query对象
             */
            QueryScorer fragmentScorer = new QueryScorer(query);
            Highlighter highlighter = new Highlighter(formatter, fragmentScorer);
            Fragmenter fragmenter = new SimpleFragmenter(100);
            /*
             * Highlighter利用Fragmenter将原始文本分割成多个片段。
             * 内置的SimpleFragmenter将原始文本分割成相同大小的片段,片段默认的大小为100个字符。这个大小是可控制的。
             */
            highlighter.setTextFragmenter(fragmenter);

            ScoreDoc[] scoreDocs = topDocs.scoreDocs;

            for (ScoreDoc scDoc : scoreDocs) {
                Document document = indexSearcher.doc(scDoc.doc);
                String id = document.get("id");
                String name = document.get("name");
                String age = document.get("age");
                String rowKey = document.get("rowKey");
                float score = scDoc.score; //相似度

                //高亮显示
                String lighterName = highlighter.getBestFragment(new IKAnalyzer(), "name", name);
                if (null == lighterName) {
                    lighterName = name;
                }

                String lighterAge = highlighter.getBestFragment(new IKAnalyzer(), "age", age);
                if (null == lighterAge) {
                    lighterAge = age;
                }

                Person person = new Person();
                person.setId(NumberUtils.toLong(id));
                person.setName(lighterName);
                person.setAge(NumberUtils.toInt(age));
                
                ids.append(id).append("\n\r");
                System.out.println(String.format("id:%s, name:%s, age:%s, rowKey:%s, 相似度:%s.", 
                        id, lighterName, lighterAge, rowKey, score));
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                indexSearcher.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }
}



2、LuceneUtil类->Lucene工具类



/**
 * lucene工具类,采用IKAnalyzer中文分词器
 * 
 * @author 林计钦
 * @version 1.0 2013-6-3 下午03:51:29
 */
public class LuceneUtil {
    /** 索引库路径 */
    private static final String indexPath = XMLPropertyConfig.getConfigXML()
            .getString("index_path");
    private static final Logger log=Logger.getLogger(LuceneUtil.class);
    
    public static IndexWriter getIndexWriter(){
        try {
            //索引库路径不存在则新建一个
            File indexFile=new File(indexPath);
            if(!indexFile.exists()) indexFile.mkdir();
            
            Directory fsDirectory = FSDirectory.open(indexFile);
            IndexWriterConfig confIndex = new IndexWriterConfig(Version.LUCENE_35, new IKAnalyzer());
            confIndex.setOpenMode(OpenMode.CREATE_OR_APPEND);
            if (IndexWriter.isLocked(fsDirectory)) {
                IndexWriter.unlock(fsDirectory);
            }
            return new IndexWriter(fsDirectory, confIndex);
        } catch (Exception e) {
            e.printStackTrace();
        }  
        return null;
    }

    /**
     * 判断索引库是否已创建
     * 
     * @return true:存在,false:不存在
     * @throws Exception
     */
    public static boolean existsIndex() throws Exception {
        File file = new File(indexPath);
        if (!file.exists()) {
            file.mkdirs();
        }
        String indexSufix = "/segments.gen";
        // 根据索引文件segments.gen是否存在判断是否是第一次创建索引
        File indexFile = new File(indexPath + indexSufix);
        return indexFile.exists();
    }
     
}




3、HBaseAPI类->HBase数据库封装类


/**
 * HBase数据库封装类
 * 
 * @author 林计钦
 * @version 1.0 2013-6-4 上午11:02:17
 */
public class HBaseAPI { 
    /**主键*/
    public static String HBASE_ROW = "row"; 
    /**列镞*/
    public static String HBASE_FAMILY = "family"; 
    /**列名*/
    public static String HBASE_QUALIFIER = "qualifier";
    /**列值*/
    public static String HBASE_QUALIFIERVALUE = "qualifiervalue";
    /**时间戳*/
    public static String HBASE_TIMESTAMP = "timestamp"; 
    
    /** 访问HBase线程池大小 */
    public static int poolSize = 1000;

    public static Configuration conf;
    private static HTablePool tablePool = null;
    private static final Logger log=Logger.getLogger(HBaseAPI.class);

    
    static {
        //来自$HBase/conf/hbase-site.xml配置文件
        conf = new Configuration();
        conf.set("hbase.master.port", XMLPropertyConfig.getConfigXML().getString("hbase.hbase_master_port"));
        conf.set("hbase.zookeeper.quorum", XMLPropertyConfig.getConfigXML().getString("hbase.hbase_zookeeper_quorum"));
        conf.set("hbase.zookeeper.property.clientPort", XMLPropertyConfig.getConfigXML().getString("hbase.hbase_zookeeper_property_clientPort"));
    }

    /**
     * HTablePool对HBase表进行CRUD操作,不推荐用HTable对HBase表进行CRUD操作。<br/><br/>
     * 
     * HTablePool可以解决HTable存在的线程不安全问题,同时通过维护固定数量的HTable对象,能够在程序运行期间复用这些HTable资源对象。
     * 
     * @return
     */
    public static HTablePool getHTablePool() {
        if (tablePool == null) {
            tablePool = new HTablePool(conf, poolSize);
        }
        return tablePool;
    }

    /**
     * 从startRow开始查询,查询maxCount条记录
     * 
     * @param tableName 表名
     * @param startRow 指定开始的行(时间戳)
     * @param maxCount 从startRow开始查询,查询maxCount条记录,最高阀值为10000
     * @return [{"timestamp":"1370412537880","id":"1","name":"zhangsan","age":"20","row":"quanzhou"}]
     */
    public static List<Map<String, String>> scan(String tableName, FilterList filterList, String startRow, int maxCount) {
        List<Map<String, String>> datas = new ArrayList<Map<String, String>>();

        ResultScanner rs = null;
        try {
            HTable table = (HTable) getHTablePool().getTable(tableName);
            Scan scan = new Scan();

            if(filterList!=null){
                scan.setFilter(filterList);
            }
            if (startRow != null && !"".equals(startRow.trim())) {
                scan.setStartRow(Bytes.toBytes(startRow));
            }
            if(maxCount<=0){
                maxCount = 10000;
            }
            if (maxCount > 10000) {
                maxCount = 10000;
            }
            scan.setCaching(maxCount + 1);
            rs = table.getScanner(scan);
            //Result类提供了raw()、list()、getValue(byte[] family, byte[] qualifier)三种方法来遍历数据
            for (Result r : rs) {
                HashMap<String, String> map = new HashMap<String, String>();
                long timestamp = 0;
                for (KeyValue kv : r.list()) {
                    timestamp = kv.getTimestamp();
                    String qualifier = Bytes.toString(kv.getQualifier()); //列名
                    String value = Bytes.toString(kv.getValue()); //列值
                    map.put(qualifier, value);

                }
                map.put(HBASE_ROW, Bytes.toString(r.getRow()));
                map.put(HBASE_TIMESTAMP, "" + timestamp);
                datas.add(map);
                // 假如到了指定条数就跳出
                if (datas.size() >= maxCount) {
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }finally{
            if(rs!=null){
                rs.close();
            }
            //table.close();
        }

        return datas;
    }
}

标签:lucene,String,org,Lucene,static,new,import,HBase
From: https://blog.51cto.com/u_2650279/6154911

相关文章

  • HBase的rowkey设计
    HBase的查询实现只提供两种方式:1、按指定RowKey获取唯一一条记录,get方法(org.apache.hadoop.hbase.client.Get)2、按指定的条件获取一批记录,scan方法(org.apache.hadoop.hbase.......
  • Couchbase Server 2.0.1企业版发布
    CouchbaseServer2.0.1企业版发布作者:chszs最近两天,CouchbaseServer2.0.1企业版发布了。正如大家预料的那样,2.0.1版是2.0版的维护版,它并没有增加很多新特征,而主要集中于......
  • CouchDB与CouchBase的比较
    《CouchDB与CouchBase的比较》作者:chszs,转载需注明。一、概述ApacheCouchDB和CouchBase两个NoSQL数据库,都是开源、免费的NoSQL文档型数据库,都使用了JSON作为其文档格式。Ap......
  • 使用HBase EndPoint(coprocessor)进行计算
    如果要统对hbase中的数据,进行某种统计,比如统计某个字段最大值,统计满足某种条件的记录数,统计各种记录特点,并按照记录特点分类(类似于sql的groupby)~常规的做法就是把hbase中整......
  • lucene4.5源码分析系列:索引缓存以及刷新
    缓存和刷新是比较重要的问题,它涉及到lucene如何管理内存和磁盘。前面提到索引的结果是缓存在内存中的,等到一定时候才会将其刷新到硬盘上去。缓存在这里的目的无非是缓解高速......
  • Lucene 源代码剖析-9索引是如何存储的
    Lucene 源码剖析5           索引是如何存储的  5.1   数据存储类Directory      Directory及相关类负责文档索引的存储。           5......
  • HBase Java API操作数据库
    场景在上面将开发环境搭建起来,要想操作操作数据库除了使用HBaseShell还可以使用JAVAAPI对HBase进行操作。注:关注公众号霸道的程序猿获取编程相关电子书、教程推送与免费......
  • Springboot 系列 (24) - Springboot+HBase 大数据存储(二)| 安装配置 Apache HBase 和 A
    ApacheHBase是Java语言编写的一款Apache开源的NoSQL型数据库,不支持SQL,不支持事务,不支持Join操作,没有表关系。ApacheHBase构建在ApacheHadoop和ApacheZoo......
  • GitHub项目Storm-HBase介绍
    ​​Storm-HBase​​​,该项目是​​TwitterStorm​​​和​​ApacheHBase​​​的结合,它使用HBasecluster作为Storm的Spout数据源,目前只是初步实现,后续会进一步完善。​......
  • 简单介绍一下HBase、Cassandra、Voldemort、Redis、VoltDB、MySQL(转)
    hbase1.简介:HBase–HadoopDatabase,是一个高可靠性、高性能、面向列、可伸缩的分布式存储系统,利用HBase技术可在廉价PCServer上搭建起大规模结构化存储集群2.HBase和R......