HBase Indexer导致Solr与HBase数据不一致问题解决

太过爱你忘了你带给我的痛 2022-08-23 00:45 116阅读 0赞

## 1. 问题描述 ##

经过对HBase Indexer一段时间的使用、测试、观察，发现通过Phoenix端导入到HBase表的数据与Solr那边的数据会产生不一致的现象，具体体现为Solr那边的数据会比HBase表数据少几千条或者更多。在公司测试环境以及试点项目生产环境下都会出现这个问题。

公司的测试环境如下描述：  
1) C\_PICRECORD表，拥有两个二级索引表C\_PICRECORD\_IDX和C\_PICRECORD\_IDX\_COLLISION  
2) 测试数据80+w条  
3) psql方式导入，发现Solr会比HBase源表少几千条到上万条

**【注意】**

这个现象使用了现场的部分数据集并且在HBase表创建了二级索引表的情况下会很容易重现。

## 2. 发现问题 ##

HBase indexer内置会有一些监控数据，监控SepEvent事件处理次数、Solr索引添加次数、Solr索引删除次数，同时还会记录一些操作失败的次数。通过官方Github上Wiki对监控数据这部分的介绍：  
[https://github.com/NGDATA/hbase-indexer/wiki/Metrics][https_github.com_NGDATA_hbase-indexer_wiki_Metrics]

我们采用JMX方式来获取这部分监控数据，具体操作如下：  
1) 在CDH Cloudera Manager管理界面里面配置Key Value Store Indexer服务打开JMX服务配置：

![hbase-indexer-jmx-config][]

配置项包括：

-Dcom.sun.management.jmxremote.port=8999
    -Dcom.sun.management.jmxremote.authenticate=false
    -Dcom.sun.management.jmxremote.ssl=false

配置保存后重启Key Value Store Indexer服务，接下来我们就可以通过JConsole连接到这个端口，来获取HBase Indexer的监控数据了。

2) 本地测试使用JConsole来获取HBase Indexer相关监控数据，如下图所示：

![jconsole-config][]

输入需要监控的机器IP及端口（前面的配置采用8999端口），点击“连接”进入监控界面。

![jconsole-show-mbean][]

在“MBean”面板下我们可以看到有“hbaseindexer”的相关监控数据，根据HBase Indexer官网Wiki介绍，在“DirectSolrInputDocumentWriter”下保存了HBase Indexer对Solr进行操作的监控数据，比如添加索引次数“Indexer adds”和删除索引次数“Indexer deletes”，里面分别记录了索引添加次数和索引删除次数。我们正是通过监控这两个数据，分析出Solr和HBase源表数据产生不一致的原因——**部分HBase添加在HBase Indexer被当成“Indexer deletes”操作处理，从而使得Solr的数据少于HBase源表数据。少掉的数据等于“Indexer deletes”里面记录的次数，即有：Hbase Row Count - Solr numfound = Indexer deletes count。**

## 3. 调试并定位问题 ##

针对上面观察并分析的现象，我们需要定位到Hbase Indexer源码进行分析：

**【注意】**

需要使用CDH5.4里面的Key Value Store Indexer源码，而不是Github上面HBase Indexer的源码。当然，两者的区别仅在于hbase版本的不同。（CDH使用的是自己封装改造过的HBase版本）

定位到Indexer类的indexRowData方法，这个方法会调用calculateIndexUpdates方法，针对我们的配置采用的是RowBasedIndexer（Indexer的子类，并且是内部类）。因此我们定位到RowBasedIndexer的calculateIndexUpdates实现方法上。

@Override
            protected void calculateIndexUpdates(List<RowData> rowDataList, SolrUpdateCollector updateCollector) throws IOException {
    
                Map<String, RowData> idToRowData = calculateUniqueEvents(rowDataList); // 选取需要处理的行(包括那些要删除的)
    
                for (RowData rowData : idToRowData.values()) {
                    String tableName = new String(rowData.getTable(), Charsets.UTF_8);
    
                    Result result = rowData.toResult(); // 对于需要delete(delete or delete family)的行,其kvs为空
    
                    if (conf.getRowReadMode() == RowReadMode.DYNAMIC) {
                        if (!mapper.containsRequiredData(result)) {
                            if(log.isDebugEnabled()) {
                                log.debug("Row " + Bytes.toString(rowData.getRow()) + " need to re-read from hbase");
                            }
                            result = readRow(rowData, result);
                        }
                    }
    
                    boolean rowDeleted = result.isEmpty(); // 由此判断是否为delete row
    
                    String documentId;
                    if (uniqueKeyFormatter instanceof UniqueTableKeyFormatter) {
                        documentId = ((UniqueTableKeyFormatter) uniqueKeyFormatter).formatRow(rowData.getRow(),
                                rowData.getTable());
                    } else {
                        documentId = uniqueKeyFormatter.formatRow(rowData.getRow());
                    }
    
                    if (rowDeleted) {
                        // Delete row from Solr as well
                        updateCollector.deleteById(documentId);
                        if (log.isDebugEnabled()) { 
                            log.debug("Row " + Bytes.toString(rowData.getRow()) + ": deleted from Solr, kvs : " + rowData.getKeyValues());
                        }
                    } else {
                        IdAddingSolrUpdateWriter idAddingUpdateWriter = new IdAddingSolrUpdateWriter(
                                conf.getUniqueKeyField(),
                                documentId,
                                conf.getTableNameField(),
                                tableName,
                                updateCollector);
                        mapper.map(result, idAddingUpdateWriter);
                    }
                }
            }

出现不一致的根源在于

conf.getRowReadMode() == RowReadMode.DYNAMIC

我们所创建的Indexer默认情况下RowReadMode是采用DYNAMIC的，那么在代码里面会进入这段条件逻辑，这部分代码简单说明如下：

判断Result里面是否包含了所有我们需要映射到Solr里面的列，如否，则需要调用readRow方法从HBase里面获取我们需要映射的数据（根据rowkey，调用HTable.get(Get)）。

出现不一致的现象就在这部分逻辑里面，我们考虑下面这种情况：

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

经过以上分析再加上我们通过添加debug日志进行调试，验证了我们上面的猜测。

![hbase-indexer-log4j-config][]

配置CDH Key Value Store Indexer log4j配置，加入需要DEBUG的类：

log4j.logger.com.ngdata.hbaseindexer.indexer.Indexer$RowBasedIndexer=DEBUG
    log4j.logger.com.ngdata.hbaseindexer.indexer.Indexer$ColumnBasedIndexer=DEBUG

我们通过在源码里面添加输出自定义DEBUG信息来调试（需要编译并替换集群上面的hbase-indexer-engine-xxx.jar）。

## 4. 解决问题 ##

根据上面的验证以及对代码的初步了解，我们给出以下两种解决方案：

### 4.1 方案1 ###

创建hbase indexer的时候，对morphline-hbase-mapper.xml里面`<indexer>`节点增加read-row属性配置，配置read-row=”never”，关于这个参数的说明参考：

[https://github.com/NGDATA/hbase-indexer/wiki/Indexer-configuration\#read-row][https_github.com_NGDATA_hbase-indexer_wiki_Indexer-configuration_read-row]

配置为never，将不会进入上面提到的条件逻辑代码，那么自然也不会出现不同步的现象。**这种解决方案只需要修改创建indexer的配置，而不需要修改HBase Indexer源代码，优于后面提到的【方案2】**

### 4.2 方案2 ###

在readRow()方法里面加入retry逻辑，合理的设置重试次数并合理休眠，以此来保证能够获取到正确的Result。这种方式相对比较麻烦，首先需要修改源代码，并进行编译，替换集群上相应jar包。而且，重试和休眠会对性能造成一定影响，**建议不到万不得已不要使用这种方式。**

[https_github.com_NGDATA_hbase-indexer_wiki_Metrics]: https://github.com/NGDATA/hbase-indexer/wiki/Metrics
[hbase-indexer-jmx-config]: http://7xtamf.com1.z0.glb.clouddn.com/image/blog/hbase-indexer-issue78/hbase-indexer-jmx-config.png
[jconsole-config]: http://7xtamf.com1.z0.glb.clouddn.com/image/blog/hbase-indexer-issue78/jconsole-config.jpg
[jconsole-show-mbean]: http://7xtamf.com1.z0.glb.clouddn.com/image/blog/hbase-indexer-issue78/jconsole-show-mbean.jpg
[hbase-indexer-log4j-config]: http://7xtamf.com1.z0.glb.clouddn.com/image/blog/hbase-indexer-issue78/hbase-indexer-log4j-config.png
[https_github.com_NGDATA_hbase-indexer_wiki_Indexer-configuration_read-row]: https://github.com/NGDATA/hbase-indexer/wiki/Indexer-configuration#read-row