Lily HBase Indexer同步HBase二级索引到Solr丢失数据的问题分析

sdfouhdso888

一、问题描述　　二、分析步骤
　　2.1 查看日志
　　2.2 修改Solr的硬提交
　　2.3 寻求StackOverFlow帮助
　　2.4 修改了read-row="never"后，丢失部分字段
　　2.5 修改代码
　　2.6 重新打包分发
　　三、结果
　　四、思考
一、问题描述
　　部分业务需要使用HBase的数据进行多维度分析，我们采用了将部分数据同步到Solr，通过Solr进行多维度查询返回对应的Rowkey，再从HBase批量获取数据。因此我们使用了一个比较成熟的方案Lily HBase Indexer来同步二级索引到Solr。但是使用的时候出现了Solr丢失数据的问题。基本上每天Solr都会比HBase少几千条数据。
二、分析步骤
　　由于我们使用的是CDH集群，下面所有操作都是基于该环境
2.1 查看日志
　　到每个节点的/var/log/hbase-solr和/var/log/solr查看了日志，都没发现写入失败的记录
2.2 修改Solr的硬提交
　　由于日志没有发现错误，猜测是Solr的数据在缓存中没提交上去。
　　在solr的collecttion目录下的conf/solrconfig.xml文件，将Solr的硬提交激活，操作如下

• <autoCommit>
  
• <maxTime>${solr.autoCommit.maxTime:60000}</maxTime>
  
• <openSearcher>true</openSearcher>
  
• </autoCommit>
  

　　然后保存配置，将修改update 到Solr集群。然后测试仍旧出现上述问题
2.3 寻求StackOverFlow帮助
　　目前是没看到问题出在哪里了，因此只能去网上搜索一下具体原因了。网上有这么两个帖子
　　hbase-indexer solr numFound different from hbase table rows>
　　HBase Indexer导致Solr与HBase数据不一致问题解决
　　他们都提到了修改morphline-hbase-mapper.xml，添加read-row
　　如下：

　　重新刷新hbase-indexer配置
　　这次发现数目对了，但是字段缺了
2.4 修改了read-row="never"后，丢失部分字段
　　由于设置了read-row之后数据不会再次从HBase中获取，因此只会读取WAL。假如修改了部分字段，HBaseIndexer就会提交相应的字段上去。例如
　　HBase中有name和age两个字段

• put 'HBase_Indexer_Test','001','cf1:name','xiaoming'
  
• put 'HBase_Indexer_Test','002','cf1:name','xiaohua'
  

　　此时的数据为

　　然后执行

• put 'HBase_Indexer_Test','001','cf1:age','12'
  

　　最后只能看到

　　说明这种模式只从WAL获取数据，并且将获取的数据覆盖到了Solr里面。
　　那么这样看来只能修改HBase indexer的代码了
2.5 修改代码
　　Lily HBase Indexer的代码是托管在github 上的，如果是单独安装的请直接访问NGDATA的这个工程：http://ngdata.github.io/hbase-indexer/
　　如果是使用的CDH版本，请访问：https://github.com/cloudera/hbase-indexer
　　我这里使用CDH 5.7.0版本进行测试。在releases选项中可以找到对应版本号的包，下载解压之后可以看到一个Maven工程。可以看到它包含如下模块

　　在./hbase-indexer-engine/src/main/java/com/ngdata/hbaseindexer/indexer/Indexer.java文件中有一个calculateIndexUpdates方法，其中有如下代码：

• Result result = rowData.toResult();
  
• if(conf.getRowReadMode()==RowReadMode.DYNAMIC){
  
• if(!mapper.containsRequiredData(result)){
  
•         result = readRow(rowData);
  
• }
  
• }
  
• 
  
• boolean rowDeleted = result.isEmpty();
  

• privateResult readRow(RowData rowData)throwsIOException{
  
• TimerContext timerContext = rowReadTimer.time();
  
• try{
  
• HTableInterface table = tablePool.getTable(rowData.getTable());
  
• try{
  
• Getget= mapper.getGet(rowData.getRow());
  
• return table.get(get);
  
• }finally{
  
•             table.close();
  
• }
  
• }finally{
  
•         timerContext.stop();
  
• }
  
• }
  

　　从代码中可以看出其执行的流程图如下：

　　假如我们使用默认的Dynamic模式写入了大量的数据，那么意味着有部分数据会在WAL生成后一段时间内无法“落地”，那么就可能出现下面的情况：

• HBase RegionServer 将Put操作先写WAL （这个时候Put还没保存到Region）；
  
• 异步处理的HBase Indexer获取到这个WAL日志，对数据进行处理，进入了我们上面说的这段条件逻辑代码，恰巧Result里面没有一部分Solr索引列，那么需要调用readRow方法从HBase重新读取数据，这个时候调用HTable.get(Get) 并没有获取到数据（Result.isEmpty()为真）；
  
• HBase RegionServer把Put保存到Region ；
  
• 那么对于第二个步骤里面的HBase Indexer，那条记录将被当成delelet操作，所以在后面的逻辑将其当成solr delete document的操作。所以在Solr中才会出现部分数据丢失和数值不对。
  

　　知道了问题在哪里之后，我们尝试修改他的源码。由于HBase将预写日志的内容写到HBase region中会有一定的滞后性，因此我们可以认为预写日志中的内容总是最新的数据。假设我们有一条rowkey =001的数据如下：
列名值Rowkey001cf1:Aacf1:Bbcf1:Cc　　我们将C的值改成D。由于夹杂在很多条数据中，可能日志中拿到了C = 'd'，但是HBase中仍旧是'c',我们需要将HBase的数据拿出来，再将预写日志中的数据覆盖它，便有了下面的代码

• privateResult readRow(RowData rowData)throwsIOException{
  
• TimerContext timerContext = rowReadTimer.time();
  
• try{
  
• HTableInterface table = tablePool.getTable(rowData.getTable());
  
• try{
  
• Get get = mapper.getGet(rowData.getRow());
  
• return merge(table.get(get), rowData.toResult());
  
• //return  table.get(get);
  
• }finally{
  
•             table.close();
  
• }
  
• }finally{
  
•         timerContext.stop();
  
• }
  
• }
  
• 
  
• privateResult merge(Result data,Result wal)throwsIOException{
  
• //如果data为空，则直接返回WAL的数据
  
• if(data.isEmpty()){
  
• return wal;
  
• }
  
• /*  //如果rowkey不相同，则返回wal的数据
  
•     if (!Bytes.toString(data.getRow()).equals(Bytes.toString(wal.getRow()))) {
  
•         return wal;
  
•     }*/
  
• TreeMap<String,Cell> cellMap =newTreeMap<String,Cell>();
  
• CellScanner dataScanner = data.cellScanner();
  
• CellScanner walScanner = wal.cellScanner();
  
• 
  
• while(dataScanner.advance()){
  
• Cell cell = dataScanner.current();
  
• String cf =Bytes.toString(CellUtil.cloneFamily(cell));
  
• String cq =Bytes.toString(CellUtil.cloneQualifier(cell));
  
• String key = cf +"->"+ cq;
  
•         cellMap.put(key, cell);
  
• }
  
• 
  
• while(walScanner.advance()){
  
• Cell cell = walScanner.current();
  
• String cf =Bytes.toString(CellUtil.cloneFamily(cell));
  
• String cq =Bytes.toString(CellUtil.cloneQualifier(cell));
  
• String key = cf +"->"+ cq;
  
•         cellMap.put(key, cell);
  
• }
  
• 
  
• ArrayList<Cell> cells =newArrayList<Cell>();
  
•     cells.addAll(cellMap.values());
  
• returnResult.create(cells);
  
• }
  

　　值得一提的是，HBase返回的result中，列的排序是按照"列族名+列名"的字典排序。比如表中有["cf1:name","cf2:cellphone","cf1:age"] 三个列，那么返回的时候会排列成["cf1:age","cf1:name","cf2:cellphone"]。在创建新的Result对象的时候也必须遵循这样的规则，因此这里使用了treemap。不要问我为什么，我特么调了一整天才发现这个问题。

2.6 重新打包分发
　　进入hbase-indexer-engine的工程，执行mvn clean install -DskipTests进行打包，稍等片刻便好了

　　在target下面有一个hbase-indexer-engine-1.5-cdh5.7.0.jar文件（这里的版本号对应自己的环境），将这个jar文件分发到集群的hbase-indexer的目录下，CDH版本放在在/opt/cloudera/parcels/CDH/jars/下即可。
　　然后重启服务进行测试。
三、结果
　　数据跑了一天，Solr中对应的条数和HBase的一样。
　　因此我们修改的代码是有效的。
四、思考
　　上面我们是合并了数据然后全部覆盖到Solr的，如果HBase存在大量的Update操作，那么势必每次列数都会和映射到Solr里面的列不一致，因此每次都会从HBase中get一次数据，这样肯定会影响性能。那么我们能否使用ReadRow.Never模式 + Solr的原子更新的方式来实现呢？