你好,游客 登录
背景:
阅读新闻

大数据架构-使用HBase和Solr配置存储与索引

[日期:2014-09-16] 来源:  作者: [字体: ]

    HBase和Solr可以通过协处理器Coprocessor的方式向Solr发出请求,Solr对于接收到的数据可以做相关的同步:增、删、改索 引的操作。将存储与索引放在不同的机器上,这是大数据架构的必须品,但目前还有很多不懂得此道的同学,他们对于这种思想感到很新奇,不过,这绝对是好的方 向,所以不懂得抓紧学习吧。

    有个朋友给我的那篇博客留言,说CDH也可以做这样的事情,我还没有试过,他还问我要与此相关的代码,于是我就稍微整理了一下,作为本篇文章的主要内容。关于CDH的事,我会尽快尝试,有知道的同学可以给我留言。

    下面我主要讲述一下,我测试对HBase和Solr的性能时,使用HBase协处理器向HBase添加数据所编写的相关代码,及解释说明。

一、编写HBase协处理器Coprocessor

    ​一旦有数据postPut,就立即对Solr里相应的Core更新。这里使用了ConcurrentUpdateSolrServer,它是Solr速率性能的保证,使用它不要忘记在Solr里面配置autoCommit哟。

  1. /* 
  2.  
  3.  *版权:王安琪 
  4.  
  5.  *描述:监视HBase,一有数据postPut就向Solr发送,本类要作为触发器添加到HBase 
  6.  
  7.  *修改时间:2014-05-27 
  8.  
  9.  *修改内容:新增 
  10.  
  11.  */ 
  12.  
  13. package solrHbase.test; 
  14.  
  15.  
  16.  
  17. import java.io.UnsupportedEncodingException; 
  18.  
  19.  
  20.  
  21. import ***; 
  22.  
  23.  
  24.  
  25. public class SorlIndexCoprocessorObserver extends BaseRegionObserver { 
  26.  
  27.  
  28.  
  29.     private static final Logger LOG = LoggerFactory 
  30.  
  31.             .getLogger(SorlIndexCoprocessorObserver.class); 
  32.  
  33.     private static final String solrUrl = "http://192.1.11.108:80/solr/core1"
  34.  
  35.     private static final SolrServer solrServer = new ConcurrentUpdateSolrServer( 
  36.  
  37.             solrUrl, 10000, 20); 
  38.  
  39.  
  40.  
  41.     /** 
  42.  
  43.      * 建立solr索引 
  44.  
  45.      * 
  46.  
  47.      * @throws UnsupportedEncodingException 
  48.  
  49.      */ 
  50.  
  51.     @Override 
  52.  
  53.     public void postPut(final ObserverContext<RegionCoprocessorEnvironment> e, 
  54.  
  55.             final Put put, final WALEdit edit, final boolean writeToWAL) 
  56.  
  57.             throws UnsupportedEncodingException { 
  58.  
  59.         inputSolr(put); 
  60.  
  61.     } 
  62.  
  63.  
  64.  
  65.     public void inputSolr(Put put) { 
  66.  
  67.         try { 
  68.  
  69.             solrServer.add(TestSolrMain.getInputDoc(put)); 
  70.  
  71.         } catch (Exception ex) { 
  72.  
  73.             LOG.error(ex.getMessage()); 
  74.  
  75.         } 
  76.  
  77.     } 
  78.  

    注意:getInputDoc是这个HBase协处理器Coprocessor的精髓所在,它可以把HBase内的Put里的内容转化成Solr需 要的值。其中String fieldName = key.substring(key.indexOf(columnFamily) + 3, key.indexOf(" 我在这")).trim();这里有一个乱码字符,在这里看不到,请大家注意一下。

  1. public static SolrInputDocument getInputDoc(Put put) { 
  2.  
  3.         SolrInputDocument doc = new SolrInputDocument(); 
  4.  
  5.         doc.addField("test_ID", Bytes.toString(put.getRow())); 
  6.  
  7.         for (KeyValue c : put.getFamilyMap().get(Bytes.toBytes(columnFamily))) { 
  8.  
  9.             String key = Bytes.toString(c.getKey()); 
  10.  
  11.             String value = Bytes.toString(c.getValue()); 
  12.  
  13.             if (value.isEmpty()) { 
  14.  
  15.                 continue; 
  16.  
  17.             } 
  18.  
  19.             String fieldName = key.substring(key.indexOf(columnFamily) + 3, 
  20.  
  21.                     key.indexOf("")).trim(); 
  22.  
  23.             doc.addField(fieldName, value); 
  24.  
  25.         } 
  26.  
  27.         return doc; 
  28.  
  29.     } 

二、编写测试程序入口代码main

    ​这段代码向HBase请求建了一张表,并将模拟的数据,向HBase连续地提交数据内容,在HBase中不断地插入数据,同时记录时间,测试插入性能。

  1. /* 
  2.  
  3.  *版权:王安琪 
  4.  
  5.  *描述:测试HBaseInsert,HBase插入性能 
  6.  
  7.  *修改时间:2014-05-27 
  8.  
  9.  *修改内容:新增 
  10.  
  11.  */ 
  12.  
  13. package solrHbase.test; 
  14.  
  15.  
  16.  
  17. import hbaseInput.HbaseInsert; 
  18.  
  19.  
  20.  
  21. import ***; 
  22.  
  23.  
  24.  
  25. public class TestHBaseMain { 
  26.  
  27.  
  28.  
  29.     private static Configuration config; 
  30.  
  31.     private static String tableName = "angelHbase"
  32.  
  33.     private static HTable table = null
  34.  
  35.     private static final String columnFamily = "wanganqi"
  36.  
  37.  
  38.  
  39.     /** 
  40.  
  41.      * @param args 
  42.  
  43.      */ 
  44.  
  45.     public static void main(String[] args) { 
  46.  
  47.         config = HBaseConfiguration.create(); 
  48.  
  49.         config.set("hbase.zookeeper.quorum", "192.103.101.104"); 
  50.  
  51.         HbaseInsert.createTable(config, tableName, columnFamily); 
  52.  
  53.         try { 
  54.  
  55.             table = new HTable(config, Bytes.toBytes(tableName)); 
  56.  
  57.             for (int k = 0; k < 1; k++) { 
  58.  
  59.                 Thread t = new Thread() { 
  60.  
  61.                     public void run() { 
  62.  
  63.                         for (int i = 0; i < 100000; i++) { 
  64.  
  65.                             HbaseInsert.inputData(table, 
  66.  
  67.                                     PutCreater.createPuts(1000, columnFamily)); 
  68.  
  69.                             Calendar c = Calendar.getInstance(); 
  70.  
  71.                             String dateTime = c.get(Calendar.YEAR) + "-" 
  72.  
  73.                                     + c.get(Calendar.MONTH) + "-" 
  74.  
  75.                                     + c.get(Calendar.DATE) + "T" 
  76.  
  77.                                     + c.get(Calendar.HOUR) + ":" 
  78.  
  79.                                     + c.get(Calendar.MINUTE) + ":" 
  80.  
  81.                                     + c.get(Calendar.SECOND) + ":" 
  82.  
  83.                                     + c.get(Calendar.MILLISECOND) + "Z 写入: " 
  84.  
  85.                                     + i * 1000; 
  86.  
  87.                             System.out.println(dateTime); 
  88.  
  89.                         } 
  90.  
  91.                     } 
  92.  
  93.                 }; 
  94.  
  95.                 t.start(); 
  96.  
  97.             } 
  98.  
  99.         } catch (IOException e1) { 
  100.  
  101.             e1.printStackTrace(); 
  102.  
  103.         } 
  104.  
  105.     } 
  106.  
  107.  
  108.  

    ​下面的是与HBase相关的操作,把它封装到一个类中,这里就只有建表与插入数据的相关代码。

  1. /* 
  2.  
  3.  *版权:王安琪 
  4.  
  5.  *描述:与HBase相关操作,建表与插入数据 
  6.  
  7.  *修改时间:2014-05-27 
  8.  
  9.  *修改内容:新增 
  10.  
  11.  */ 
  12.  
  13. package hbaseInput; 
  14.  
  15. import ***; 
  16.  
  17. import org.apache.hadoop.hbase.client.Put; 
  18.  
  19.  
  20.  
  21. public class HbaseInsert { 
  22.  
  23.  
  24.  
  25.     public static void createTable(Configuration config, String tableName, 
  26.  
  27.             String columnFamily) { 
  28.  
  29.         HBaseAdmin hBaseAdmin; 
  30.  
  31.         try { 
  32.  
  33.             hBaseAdmin = new HBaseAdmin(config); 
  34.  
  35.             if (hBaseAdmin.tableExists(tableName)) { 
  36.  
  37.                 return; 
  38.  
  39.             } 
  40.  
  41.             HTableDescriptor tableDescriptor = new HTableDescriptor(tableName); 
  42.  
  43.             tableDescriptor.addFamily(new HColumnDescriptor(columnFamily)); 
  44.  
  45.             hBaseAdmin.createTable(tableDescriptor); 
  46.  
  47.             hBaseAdmin.close(); 
  48.  
  49.         } catch (MasterNotRunningException e) { 
  50.  
  51.             e.printStackTrace(); 
  52.  
  53.         } catch (ZooKeeperConnectionException e) { 
  54.  
  55.             e.printStackTrace(); 
  56.  
  57.         } catch (IOException e) { 
  58.  
  59.             e.printStackTrace(); 
  60.  
  61.         } 
  62.  
  63.     } 
  64.  
  65.  
  66.  
  67.     public static void inputData(HTable table, ArrayList<Put> puts) { 
  68.  
  69.         try { 
  70.  
  71.             table.put(puts); 
  72.  
  73.             table.flushCommits(); 
  74.  
  75.             puts.clear(); 
  76.  
  77.         } catch (IOException e) { 
  78.  
  79.             e.printStackTrace(); 
  80.  
  81.         } 
  82.  
  83.     } 
  84.  

三、编写模拟数据Put

    向HBase中写入数据需要构造Put,下面是我构造模拟数据Put的方式,有字符串的生成,我是由mmseg提供的词典words.dic中随机读取一些词语连接起来,生成一句字符串的,下面的代码没有体现,不过很easy,你自己造你自己想要的数据就OK了。

  1. public static Put createPut(String columnFamily) { 
  2.  
  3.         String ss = getSentence(); 
  4.  
  5.         byte[] family = Bytes.toBytes(columnFamily); 
  6.  
  7.         byte[] rowKey = Bytes.toBytes("" + Math.abs(r.nextLong())); 
  8.  
  9.         Put put = new Put(rowKey); 
  10.  
  11.         put.add(family, Bytes.toBytes("DeviceID"), 
  12.  
  13.                 Bytes.toBytes("" + Math.abs(r.nextInt()))); 
  14.  
  15.         ****** 
  16.  
  17.         put.add(family, Bytes.toBytes("Company_mmsegsm"), Bytes.toBytes("ss")); 
  18.  
  19.  
  20.  
  21.         return put; 
  22.  
  23.     } 

    当然在运行上面这个程序之前,需要先在Solr里面配置好你需要的列信息,HBase、Solr安装与配置,它们的基础使用方法将会在之后的文章中 介绍。在这里,Solr的列配置就跟你使用createPut生成的Put搞成一样的列名就行了,当然也可以使用动态列的形式。

四、直接对Solr性能测试

    如果你不想对HBase与Solr的相结合进行测试,只想单独对Solr的性能进行测试,这就更简单了,完全可以利用上面的代码段来测试,稍微组装一下就可以了。

  1. private static void sendConcurrentUpdateSolrServer(final String url, 
  2.  
  3.             final int count) throws SolrServerException, IOException { 
  4.  
  5.         SolrServer solrServer = new ConcurrentUpdateSolrServer(url, 10000, 20); 
  6.         for (int i = 0; i < count; i++) {      solrServer.add(getInputDoc(PutCreater.createPut(columnFamily))); 
  7.         } 
  8.     } 

    希望可以帮助到你规格严格-功夫到家。这次的文章代码又偏多了点,但代码是解释思想的最好的语言,我的提倡就是尽可能的减少代码的注释,尽力简化你的代码,使你的代码足够的清晰易懂,甚至于相似于伪代码了,这也是《重构》这本书里所提倡的。

原文链接:http://www.cnblogs.com/wgp13x/p/3927979.html





收藏 推荐 打印 | 录入: | 阅读:
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数
点评:
       
评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款