HBase协处理器同步二级索引到Solr

一、 背景
二、 什么是HBase的协处理器
三、 HBase协处理器同步数据到Solr
四、 添加协处理器
五、 测试
六、 协处理器动态加载

一、 背景

在实际生产中，HBase往往不能满足多维度分析，我们能想到的办法就是通过创建HBase数据的二级索引来快速获取rowkey，从而得到想要的数据。目前比较流行的二级索引解决方案有Lily HBase Indexer，Phoenix自带的二级索引，华为Indexer，以及360的二级索引方案。上面的目前使用比较广泛的应该是Lily HBase Indexer，但是我们有时候只想实现一些简单的功能或者比较特殊的功能的时候，需要自己写协处理器进行处理。学习HBase的协处理器对于了解HBase架构是有帮助的。

二、 什么是HBase的协处理器

协处理器分两种类型，系统协处理器可以全局导入region server上的所有数据表，表协处理器即是用户可以指定一张表使用协处理器。

Hbase的coprocessor分为两类，Observer和EndPoint。其中Observer相当于触发器，EndPoint相当于存储过程。其中Observer的代码部署在服务端，相当于对API调用的代理。

另一个是终端(endpoint)，动态的终端有点像存储过程。

Observer

观察者的设计意图是允许用户通过插入代码来重载协处理器框架的upcall方法，而具体的事件触发的callback方法由HBase的核心代码来执行。协处理器框架处理所有的callback调用细节，协处理器自身只需要插入添加或者改变的功能。以HBase0.92版本为例，它提供了三种观察者接口：

  • RegionObserver：提供客户端的数据操纵事件钩子：Get、Put、Delete、Scan等。
  • WALObserver：提供WAL相关操作钩子。
  • MasterObserver：提供DDL-类型的操作钩子。如创建、删除、修改数据表等。

这些接口可以同时使用在同一个地方，按照不同优先级顺序执行.用户可以任意基于协处理器实现复杂的HBase功能层。HBase有很多种事件可以触发观察者方法，这些事件与方法从HBase0.92版本起，都会集成在HBase API中。不过这些API可能会由于各种原因有所改动，不同版本的接口改动比较大。

三、 HBase协处理器同步数据到Solr

实时更新数据需要获取到HBase的插入、更新和删除操作。由于HBase中的插入和更新都是对应RegionServer的Put操作，因此我们需要使用RegionObserver中的"postPut"和"postDelete函数"。至于Truncate操作则需要使用MasterObserver。

我们需要做的就是拦截put和delete操作，将里面的内容获取出来，写入Solr。 对应的协处理器代码如下：

    

  1. package com.bqjr.bigdata.HBaseObserver.server;
    2. 

    3. 

  3. import com.bqjr.bigdata.HBaseObserver.entity.SolrServerManager;
    4. 

  4. import org.apache.hadoop.hbase.CellUtil;
    5. 

  5. import org.apache.hadoop.hbase.client.Delete;
    6. 

  6. import org.apache.hadoop.hbase.client.Durability;
    7. 

  7. import org.apache.hadoop.hbase.client.Put;
    8. 

  8. import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
    9. 

  9. import org.apache.hadoop.hbase.coprocessor.ObserverContext;
    10. 

  10. import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
    11. 

  11. import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
    12. 

  12. import org.apache.hadoop.hbase.util.Bytes;
    13. 

  13. import org.apache.solr.client.solrj.SolrServerException;
    14. 

  14. import org.apache.solr.common.SolrInputDocument;
    15. 

  15. import java.io.IOException;
    16. 

  16. /**
    17. 

  17.  * Created by hp on 2017-02-15. */
    18. 

  18.  public class HBaseIndexerToSolrObserver extends BaseRegionObserver{
    19. 

  19.     String[] columns = {"test_age","test_name"};
    20. 

  20.   String collection = "bqjr";
    21. 

  21.   SolrServerManager solrManager = new SolrServerManager(collection);
    22. 

    23. 

  23.   @Override
    24. 

  24.   public void postPut(ObserverContext<RegionCoprocessorEnvironment> e,
    25. 

  25.   Put put, WALEdit edit, Durability durability) throws IOException {
    26. 

  26.         String rowkey= Bytes.toString(put.getRow());
    27. 

    28. 

  28.   SolrInputDocument doc = new SolrInputDocument();
    29. 

  29.  for(String column : columns){
    30. 

  30.  if(put.has(Bytes.toBytes("cf1"),Bytes.toBytes(column))){
    31. 

  31.  doc.addField(column,Bytes.toString(CellUtil.cloneValue(put.get(Bytes.toBytes("cf1"),Bytes.toBytes(column)).get(0))));
    32. 

  32. }
    33. 

  33.   }
    34. 

  34.         try {
    35. 

  35.             solrManager.addDocToCache(doc);
    36. 

  36.   } catch (SolrServerException e1) {
    37. 

  37.             e1.printStackTrace();
    38. 

  38.   }
    39. 

    40. 

  40.     }
    41. 

    42. 

  42.     @Override
    43. 

  43.   public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e,
    44. 

  44.   Delete delete,
    45. 

  45.   WALEdit edit,
    46. 

  46.   Durability durability) throws IOException{
    47. 

  47.         String rowkey= Bytes.toString(delete.getRow());
    48. 

  48.  try {
    49. 

  49.             solrManager.delete(rowkey);
    50. 

  50.   } catch (SolrServerException e1) {
    51. 

  51.             e1.printStackTrace();
    52. 

  52.   }
    53. 

  53.     }
    54. 

    55. 

  55. }
    56.

大体的写入流程我们已经完成了，接下来就是Solr的写入实现了。由于Solr需要使用Zookeeper等信息，我们可以直接通过HBase的conf中获取Zookeeper相关信息来构造所需要的SolrCloudServer。

另一方面，我们不能来了一条数据就马上写入，这样非常消耗资源。因此我们需要做一个缓存，将这些Solr数据暂时保存在里面，定时 + 定量的发送。代码如下

    

  1. package com.bqjr.bigdata.HBaseObserver.entity;
    2. 

    3. 

  3. import org.apache.hadoop.conf.Configuration;
    4. 

  4. import org.apache.hadoop.hbase.HBaseConfiguration;
    5. 

  5. import org.apache.solr.client.solrj.SolrServerException;
    6. 

  6. import org.apache.solr.client.solrj.impl.CloudSolrServer;
    7. 

  7. import org.apache.solr.client.solrj.response.UpdateResponse;
    8. 

  8. import org.apache.solr.common.SolrInputDocument;
    9. 

    10. 

  10. import java.io.IOException;
    11. 

  11. import java.util.*;
    12. 

    13. 

  13. /**
    14. 

  14.  * Created by hp on 2017-02-15. */public class SolrServerManager {
    15. 

  15.     public static String ZKHost = "";
    16. 

  16.  public static String ZKPort = "";
    17. 

  17.  int zkClientTimeout = 1800000;// 心跳
    18. 

  18.   int zkConnectTimeout = 1800000;// 连接时间
    19. 

  19.   CloudSolrServer solrServer;
    20. 

  20.  private static String defaultCollection;
    21. 

  21.  int maxCache = 10000;
    22. 

  22.  public static List<SolrInputDocument> cache = new LinkedList<SolrInputDocument>();
    23. 

  23.  private static int maxCommitTime = 60; //最大提交时间，s
    24. 

    25. 

  25.   public SolrServerManager(String collection) {
    26. 

  26.         defaultCollection = collection;
    27. 

  27.   Configuration conf = HBaseConfiguration.create();
    28. 

  28.   ZKHost = conf.get("hbase.zookeeper.quorum", "bqdpm1,bqdpm2,bqdps2");
    29. 

  29.   ZKPort = conf.get("hbase.zookeeper.property.clientPort", "2181");
    30. 

  30.   String SolrUrl = ZKHost + ":" + ZKPort + "/" + "solr";
    31. 

  31.   solrServer = new CloudSolrServer(SolrUrl);
    32. 

  32.   solrServer.setDefaultCollection(defaultCollection);
    33. 

  33.   solrServer.setZkClientTimeout(zkClientTimeout);
    34. 

  34.   solrServer.setZkConnectTimeout(zkConnectTimeout);
    35. 

  35.   //启动定时任务，第一次延迟10执行,之后每隔指定时间执行一次
    36. 

  36.   Timer timer = new Timer();
    37. 

  37.   timer.schedule(new CommitTimer(), 10 * 1000L, maxCommitTime * 1000L);
    38. 

  38.   }
    39. 

    40. 

  40.     public UpdateResponse put(SolrInputDocument doc) throws IOException, SolrServerException {
    41. 

  41.         solrServer.add(doc);
    42. 

  42.  return solrServer.commit(false, false);
    43. 

  43.   }
    44. 

    45. 

  45.     public UpdateResponse put(List<SolrInputDocument> docs) throws IOException, SolrServerException {
    46. 

  46.         solrServer.add(docs);
    47. 

  47.  return solrServer.commit(false, false);
    48. 

  48.   }
    49. 

    50. 

  50.     public void addDocToCache(SolrInputDocument doc) throws IOException, SolrServerException {
    51. 

  51.         synchronized (cache) {
    52. 

  52.             cache.add(doc);
    53. 

  53.  if (cache.size() >= maxCache) {
    54. 

  54.                 this.put(cache);
    55. 

  55.   cache.clear();
    56. 

  56.   }
    57. 

  57.         }
    58. 

  58.     }
    59. 

    60. 

  60.     public UpdateResponse delete(String rowkey) throws IOException, SolrServerException {
    61. 

  61.         solrServer.deleteById(rowkey);
    62. 

  62.  return solrServer.commit(false, false);
    63. 

  63.   }
    64. 

    65. 

  65.     /**
    66. 

  66.  * 提交定时器 */  static class CommitTimer extends TimerTask {
    67. 

  67.         @Override
    68. 

  68.   public void run() {
    69. 

  69.             synchronized (cache) {
    70. 

  70.                 try {
    71. 

  71.                     new SolrServerManager(defaultCollection).put(cache);
    72. 

  72.   cache.clear();
    73. 

  73.   } catch (IOException e) {
    74. 

  74.                     e.printStackTrace();
    75. 

  75.   } catch (SolrServerException e) {
    76. 

  76.                     e.printStackTrace();
    77. 

  77.   }
    78. 

  78.                 cache.clear();
    79. 

  79.   }
    80. 

  80.         }
    81. 

  81.     }
    82. 

  82. }
    83.

四、 添加协处理器

    

  1. #先禁用这张表
    2. 

  2. disable ‘HBASE_OBSERVER_TEST‘
    3. 

  3. #为这张表添加协处理器,设置的参数具体为： jar文件路径|类名|优先级（SYSTEM或者USER）
    4. 

  4. alter ‘HBASE_OBSERVER_TEST‘,‘coprocessor‘=>‘hdfs://bqdpm1:8020/ext_lib/HBaseObserver-1.0.0.jar|com.bqjr.bigdata.HBaseObserver.server.HBaseIndexerToSolrObserver||‘
    5. 

  5. #启用这张表
    6. 

  6. enable ‘HBASE_OBSERVER_TEST‘
    7. 

  7. #删除某个协处理器，"$<bumber>"后面跟的ID号与desc里面的ID号相同
    8. 

  8. alter ‘HBASE_OBSERVER_TEST‘,METHOD=>‘table_att_unset‘,NAME => ‘coprocessor$1‘
    9.

五、 测试

尝试插入一条数据put ‘HBASE_OBSERVER_TEST‘,‘001‘,‘cf1:test_age‘,‘18‘

结果Solr中一条数据都没有

然后查看了regionserver的日志发现，没有找到SolrJ的类

然后我们将所有的依赖加到Jar包里面之后，再次运行。就可以看到数据了。

测试Delete功能

测试进行到这里就完了吗？当然不是

我们尝试再插入一条put ‘HBASE_OBSERVER_TEST‘,‘001‘,‘cf1:test_name‘,‘Bob‘

理论上我们需要在Solr中看到 test_age = 18,test_name = Bob。

但是在Solr中只有一条数据

于是我们需要使用到Solr的原子更新功能。将postPut改成下面这样的代码即可

    

  1. public void postPut(ObserverContext<RegionCoprocessorEnvironment> e,
    2. 

  2.   Put put, WALEdit edit, Durability durability) throws IOException {
    3. 

  3.         String rowkey= Bytes.toString(put.getRow());
    4. 

  4.   Long version = 1L;
    5. 

  5.   SolrInputDocument doc = new SolrInputDocument();
    6. 

  6.  for(String column : columns){
    7. 

  7.             if(put.has(Bytes.toBytes("cf1"),Bytes.toBytes(column))){
    8. 

  8.                 Cell cell = put.get(Bytes.toBytes("cf1"),Bytes.toBytes(column)).get(0);
    9. 

  9.   Map<String, String > operation = new HashMap<String,String>();
    10. 

  10.   operation.put("set",Bytes.toString(CellUtil.cloneValue(cell)));
    11. 

  11.   doc.setField(column,operation);
    12. 

  12.   }
    13. 

  13.         }
    14. 

  14.         doc.addField("id",rowkey);
    15. 

  15. //        doc.addField("_version_",version);
    16. 

  16.   try {
    17. 

  17.             solrManager.addDocToCache(doc);
    18. 

  18.   } catch (SolrServerException e1) {
    19. 

  19.             e1.printStackTrace();
    20. 

  20.   }
    21. 

    22. 

  22.     }
    23.

再次插入数据

查看Solr结果

六、 协处理器动态加载

hbase的官方文档指出动态级别的协处理器，可以做到不重启hbase，更新协处理，做法就是

禁用表，卸载协处理器，重新指定协处理器， 激活表，即可，但实际测试发现

动态加载无效，是hbase的一个bug，看这个链接：

https://issues.apache.org/jira/browse/HBASE-8445

因为协处理器，已经被JVM加载，即使删除jar也不能重新load的jar，因为cache里面的hdfs的jar路径，没有变化，所以动态更新无效

，除非重启JVM，那样就意味着，需要重启RegionServer，

里面的小伙伴们指出了两种办法，使协处理器加载生效：

（1）滚动重启regionserver，避免停掉所有的节点

（2）改变协处理器的jar的类名字或者hdfs加载路径，以方便有新的ClassLoad去加载它

但总体来看，第2种方法，比较安全，第一种风险太大，一般情况下没有人会随便滚动重启线上的服务器的，这只在hbase升级的时候使用

时间： 2024-11-05 22:39:52

HBase协处理器同步二级索引到Solr的相关文章

HBase 二级索引和备用查询路径

感谢平台分享-http://bjbsair.com/2020-04-10/tech-info/53319.html 你也可以将本文的标题理解为"如果我的表 rowkey 看起来像这样,但我也希望我的查询表这样." dist-list 上的一个常见示例是 row-key 格式为"user-timestamp"格式,但对于特定时间范围内的用户活动有报告要求.因此,用户选择容易,因为它处于密钥的主导位置,但时间不是. 没有一个最好的方法来解决这个问题的答案,因为它取决于:

HBase二级索引与Join

转自:http://www.oschina.net/question/12_32573 二级索引与索引Join是Online业务系统要求存储引擎提供的基本特性.RDBMS支持得比较好,NOSQL阵营也在摸索着符合自身特点的最佳解决方案.这篇文章会以HBase做为对象来探讨如何基于Hbase构建二级索引与实现索引join.文末同时会列出目前已知的包括0.19.3版secondary index, ITHbase, Facebook和官方Coprocessor方案的介绍. 理论目标在HBase中实现

hbase二级索引

二级索引与索引Join是多数业务系统要求存储引擎提供的基本特性,RDBMS早已支持,NOSQL阵营也在摸索着符合自身特点的最佳解决方案.这篇文章会以HBase做为对象来讨论如何基于Hbase构建二级索引与实现索引join.文末同时会列出目前已知的包括0.19.3版secondary index, ITHbase, Facebook方案和官方Coprocessor的介绍. 理论目标在HBase中实现二级索引与索引Join需要考虑三个目标:1,高性能的范围检索.2,数据的低冗余(存储所占的数据量).

（转）HBase二级索引与Join

二级索引与索引Join是Online业务系统要求存储引擎提供的基本特性.RDBMS支持得比较好,NOSQL阵营也在摸索着符合自身特点的最佳解决方案.这篇文章会以HBase做为对象来探讨如何基于Hbase构建二级索引与实现索引join.文末同时会列出目前已知的包括0.19.3版secondary index,?ITHbase, Facebook和官方Coprocessor方案的介绍. 理论目标在HBase中实现二级索引与索引Join需要考虑三个目标:1,高性能的范围检索.2,数据的低冗余(存储所占

HBase二级索引方案总结

转自:http://blog.sina.com.cn/s/blog_4a1f59bf01018apd.html 附hbase如何创建二级索引以及创建二级索引实例:http://www.aboutyun.com/thread-8857-1-1.html 华为二级索引(原理):http://my.oschina.net/u/923508/blog/413129 在HBase中,表格的Rowkey按照字典排序,Region按照RowKey设置split point进行shard,通过这种方式实现的全局

基于solr实现hbase的二级索引

一.目的 了解hbase的都知道,由于hbase基于行健有序存储,在查询时使用行健十分高效,然后想要实现关系型数据库那样可以随意组合的多条件查询.查询总记录数.分页等就比较麻烦了.想要实现这样的功能,我们可以采用两种方法: 使用hbase提供的filter, 自己实现二级索引,通过二级索引查询多符合条件的行健,然后再查询hbase. 第一种方法不多说了,使用起来很方便,但是局限性也很大,hbase的filter是直接扫记录的,如果数据范围很大,会导致查询速度很慢.所以如果能先使用行健把记录缩小到

CDH Solr Hbase二级索引

基于key-Value store indexer ,solrcloud创建Hbase二级索引 首先安装solrcloud,在cloudera manager 上添加solr服务,然后添加key-Value store indexer服务. 首先设置HBASE表的列族REPLICATION_SCOP =>1 如: disable ‘cloud’ alter 'cloud' ,{NAME => 'datainfo',REPLICATION_SCOPE =>'1'} enable ‘clou

【转】华为HBase索引模块应用：HBase二级索引模块：hindex调研 2014年10月16日

文章出处:http://www.batchfile.cn/?p=63 HBase二级索引模块:hindex调研 hindx是HBase的二级索引方案,为HBase提供声明式的索引,使用协处理器对索引表进行自动创建和维护,客户端不需要对数据进行双写.并且hindex采用了一些巧妙的Rowkey编排方式,使索引数据和实际数据分布在同一个Region,实现了较高的查询性能.介绍如下:huawei-hbase-secondary-secondary-index-implementations 代码下载地

HBase建立二级索引的一些解决方案

HBase的一级索引就是rowkey,我们只能通过rowkey进行检索.如果我们相对hbase里面列族的列列进行一些组合查询,就需要采用HBase的二级索引方案来进行多条件的查询. 常见的二级索引方案有以下几种: 1.MapReduce方案 2.ITHBASE方案 3.IHBASE方案 4.Coprocessor方案 5.Solr+hbase方案 MapReduce方案 IndexBuilder:利用MR的方式构建Index 优点:并发批量构建Index 缺点:不能实时构建Index 举例: 原