Storm-hbase bolt 故障转移 DRPC答案

【问题标题】：Storm-hbase bolt failing over DRPCStorm-hbase bolt 故障转移 DRPC
【发布时间】：2016-10-13 08:56:24
【问题描述】：

我正在编写一个 Storm 拓扑来使用 DRPC 从 HBase 读取数据。本质上，这会执行扫描以获取数据、丰富数据并将其返回。

我可以轻松获得一个基本的 DRPC 示例（基于 http://storm.apache.org/releases/current/Distributed-RPC.html）。但是，当我插入扫描代码时，该过程需要很长时间。一分钟后，我收到以下错误：

backtype.storm.generated.DRPCExecutionException

在 backtype.storm.daemon.drpc$service_handler$reify__8688.failRequest(drpc.clj:136) ~[storm-core-0.10.0.2.4.2.0-258.jar:0.10.0.2.4.2.0- 258]

在 backtype.storm.drpc.DRPCSpout.fail(DRPCSpout.java:241) ~[storm-core-0.10.0.2.4.2.0-258.jar:0.10.0.2.4.2.0-258

不久之后，我得到了 org.apache.hadoop.hbase.client.RetriesExhaustedException。这并不总是发生，但很常见。我基于此的假设是以下两种可能性之一：

扫描超时。但是通过 HBase Shell 或 REST 执行扫描会在不到一秒的时间内返回表不一致，导致缺少某个区域。我已经运行了 hbase hbck，它显示 0 个不一致。我知道与 HBase 的连接很好：我添加了调试输出，并且螺栓得到了结果。但是由于 DRPCExecutionException，这些结果永远不会通过 DRPC 返回。

我虽然问题是 DRPC 超时，但是我已经增加了很多 DRPC 超时，并且在相同的时间内得到了相同的结果。谷歌搜索后，我发现其他人有同样的问题 ([Storm][DRPC] Request failed)，但没有迹象表明如何解决这个问题。

作为参考，我在下面添加我的代码：

try (Table table = HbaseClient.connection().getTable(TableName.valueOf("EPG_URI"))) 
    {
        List<Filter> filters = new ArrayList<>();
        String startRowString = "start";
        String endRowString = "end";
        RowFilter startRow = new RowFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL, new BinaryPrefixComparator(startRowString.getBytes()));
        filters.add(startRow);
        RowFilter endRow = new RowFilter(CompareFilter.CompareOp.LESS_OR_EQUAL, new BinaryPrefixComparator(endRowString.getBytes()));
        filters.add(endRow);
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, filters);

        Scan scan = new Scan();
        scan.addFamily("f1".getBytes());
        scan.setFilter(filterList);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) 
        {
            hbaseValues.add(result);
        }
    }
}

提前感谢您的帮助。

【问题讨论】：

标签： hbase apache-storm

【解决方案1】：

好的，所以我不知道为什么会出现最初的问题，但我确实通过大大简化我的扫描仪来解决它：

Scan scan = new Scan(startRowString.getBytes(), endRowString.getBytes());
scan.addFamily("f1".getBytes());
ResultScanner scanner = table.getScanner(scan);            
for (Result r : scanner)
{...}

基于此，问题似乎出在 HBase 端而不是 Storm 端。

【讨论】：