HBase 统计生产集群大表行数
HBase 统计生产集群大表行数
背景: 对于其他数据存储系统来说，统计表的行数是再基本不过的操作了，一般实现都非常简单；但对于HBase这种key-value存储结构的列式数据库，统计 RowCount 的方法却有好几种不同的花样，并且执行效率差别巨大！以下来比较几种方式实现。
-
测试集群: apache-hbase-1.2.5-hadoop-2.7.2 四台服务器
-
注：以下4种方法效率依次提高
1. hbase shell 的count命令
这是最简单直接的操作，但是执行效率非常低，适用于百万级以下的小表RowCount统计！
在这里插入图片描述
hbase> count 'ns1:t1'
hbase> count 't1'
hbase> count 't1', INTERVAL => 100000
hbase> count 't1', CACHE => 1000
hbase> count 't1', INTERVAL => 10, CACHE => 1000此操作可能需要很长时间, 来运行计数 MapReduce 作业, 默认情况下每1000行显示当前计数, 计数间隔可自行指定。
默认情况下在计数扫描上启用缓存, 默认缓存大小为10行。行数为 3000W 的表测试结果：
hbase(main):001:0> count 'safeclound.tb_ammeter'结果: 默认 INTERVAL 为 1000 行时花了80分钟.
hbase(main):001:0> count 'safeclound.tb_ammeter', INTERVAL => 1000000结果: 默认 INTERVAL 为 1000000 行时约耗时 80 分钟.
2. scan方式设置过滤器循环计数 (Java 实现)
这种方式是通过添加 FirstKeyOnlyFilter 过滤器的scan进行全表扫描，循环计数 RowCount，速度较慢！ 但快于第一种count方式，基本代码如下：
public void rowCountByScanFilter(String tablename){
long rowCount = 0;
try {
// 计时
StopWatch stopWatch = new StopWatch();
stopWatch.start();
TableName name=TableName.valueOf(tablename);
// connection为类静态变量
Table table = connection.getTable(name);
Scan scan = new Scan();
// FirstKeyOnlyFilter 只会取得每行数据的第一个kv，提高count速度
scan.setFilter(new FirstKeyOnlyFilter());
ResultScanner rs = table.getScanner(scan);
for (Result result : rs) {
rowCount += result.size();
}
stopWatch.stop();
System.out.println("RowCount: " + rowCount);
System.out.println("统计耗时：" +stopWatch.getTotalTimeMillis());
} catch (Throwable e) {
e.printStackTrace();
}
}结果: 耗时约 45 分钟.
3. 利用hbase.RowCounter包执行MR任务
这种方式效率非常高！利用了hbase jar中自带的统计行数的工具类。通过 $HBASE_HOME/bin/hbase 命令执行：
hbase org.apache.hadoop.hbase.mapreduce.RowCounter 'safeclound.tb_ammeter'在这里插入图片描述
在这里插入图片描述
耗时1m40s，速度较上面两种有了质的飞跃！
4. 利用Hbase协处理器Coprocessor (Java 实现)
这是我目前发现效率最高的RowCount统计方式，利用了HBase高级特性：协处理器！
我们往往使用过滤器来减少服务器端通过网络返回到客户端的数据量。但HBase中还有一些特性让用户甚至可以把一部分计算也移动到数据的存放端，那就是协处理器 (coprocessor)。
协处理器简介: 节选自《HBase权威指南》
使用客户端API，配合筛选机制，例如，使用过滤器或限制列族的范围，都可以控制被返回到客户端的数据量。如果可以更进一步优化会更好，例如，数据的处理流程直接放到服务器端执行，然后仅返回一个小的处理结果集。这类似于一个小型的MapReduce框架，该框架将工作分发到整个集群。
协处理器允许用户在region服务器上运行自己的代码，更准确地说是允许用户执行region级的操作，并且可以使用与RDBMS中触发器（trigger）类似的功能。在客户端，用户不用关心操作具体在哪里执行，HBase的分布式框架会帮助用户把这些工作变得透明。
- 核心代码：
public void rowCountByCoprocessor(String tablename){
try {
// 提前创建connection和conf
Admin admin = connection.getAdmin();
TableName name=TableName.valueOf(tablename);
// 先 disable 表, 添加协处理器后再enable表
admin.disableTable(name);
HTableDescriptor descriptor = admin.getTableDescriptor(name);
String coprocessorClass = "org.apache.hadoop.hbase.coprocessor.AggregateImplementation";
if (! descriptor.hasCoprocessor(coprocessorClass)) {
descriptor.addCoprocessor(coprocessorClass);
}
admin.modifyTable(name, descriptor);
admin.enableTable(name);
// 计时
StopWatch stopWatch = new StopWatch();
stopWatch.start();
Scan scan = new Scan();
AggregationClient aggregationClient = new AggregationClient(conf);
System.out.println("RowCount: " + aggregationClient.rowCount(name, new LongColumnInterpreter(), scan));
stopWatch.stop();
System.out.println("统计耗时：" +stopWatch.getTotalTimeMillis());
} catch (Throwable e) {
e.printStackTrace();
}
}结果: 约耗时 43 秒, 为什么利用协处理器后速度会如此之快？
Table注册了Coprocessor之后，在执行AggregationClient的时候，会将RowCount分散到Table的每一个Region上，Region内RowCount的计算，是通过RPC执行调用接口，由Region对应的RegionServer执行InternalScanner进行的。
因此，性能的提升有两点原因:
- a. 分布式统计。将原来客户端按照Rowkey的范围单点进行扫描，然后统计的方式，换成了由所有Region所在RegionServer同时计算的过程。
- b. 使用了在RegionServer内部执行使用了InternalScanner。这是距离实际存储最近的Scanner接口，存取更加快捷。
5. FAQ
-
5.1 执行
hbase org.apache.hadoop.hbase.mapreduce.RowCounter 'safeclound.tb_ammeter'统计大表任务，结果导致其他任务提交超时、已有的 spark-streaming 任务 hold 住了？- 排查&解决:
- 首先从yarn控制台查看那些任务提交了未执行和一直执行没结束(hold) http://emr-header-1:8088/cluster/apps/ACCEPTED
- 找到长时间 hold 住的任务ID，杀掉:
yarn application -list yarn application -kill application_1562735031321_31123