hello hbase

前言
近半年本人主要在倒腾数据，遇到海量数据去重的难题，曾经尝试过各种hivesql，然而随着数据量逐渐增大，处理耗时也越来越长，各种方案一一破产。2012年11月份提过使用HBase唯一主键的方案，随即做了相关预研（参看hive&hbase解决方案测评）。该方案由于HBase转化成hive表性能问题而搁浅。但在测评报告最后的总结中提到：或许我们可以选择数据“冷热”、以及部分字段切表来优化。

2013年3月，我们在业务上做了调整，通过部分字段来区分数据的“冷热”，从此希望的烟火重新燃起……

表设计
为区分数据的“冷热”，我们采用分表方式，即冷数据A和热数据B存储于HBase的不同表，另外每天增量数据的冷数据C也有一张表来存储。相对而言，B、C两张表数据量不大，顶多上千万条记录，A才是可能成为海量（TB、PB级别）的数据表。3张表的设计缺点是增量数据导入时业务复杂，需要经过多次判断。

导入
数据导入HBase及hive时遇到了几个问题，一一道来。

数据在数量级别上分两种，一种是初始化大数据量数据，另一种是常态化增量小数据量数据。两种数据导入HBase时需要采取不同的策略。

初始化数据
我们有3亿+的初始化数据，对于海量数据存储大吞吐量写入的HBase来说是小case（根据各种测评得出的结论）。原先采用MapRedue按行读取hdfs文件数据，直接put至HBase表中方式。该方式是很普遍很通用的方式，按理说应该能顺利进行，但在我们真正运行时却超级缓慢。这里我们忽视了几个重要的问题：
1.我们数据有40+个字段，而HBase导入测评字段数少。
2.HBase表没有预先创建分区，数据在插入过程中HBase会不停的执行split操作。
3.HBase的单个regionserver适合存储几个region？
网上有不少HBase数据导入调优的资料，海量数据导入最优的一种方式是将数据转化成HBase认知的HFile形式。该方式可以通过MapReduce输出HFile

job.setOutputFormatClass(HFileOutputFormat.class);
HFileOutputFormat.configureIncrementalLoad(job, new HTable(conf, hbase_table_name));
HFileOutputFormat.setOutputPath(job, new Path(outpath));
然后通过LoadIncrementalHFiles

Configuration HBASE_CONFIG = new Configuration();
conf = HBaseConfiguration.create(HBASE_CONFIG);
SchemaMetrics.configureGlobally(conf);
String[] params = new GenericOptionsParser(conf, args).getRemainingArgs();
if (params.length != 2) {
    System.err.println("Usage: LoadHFileToHBase <hdfs_path> <hbase_table>");
    System.exit(-1);
}
LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
HTable ht=new HTable(conf,params[1]);
ht.setAutoFlush(false);
loader.doBulkLoad(new Path(params[0]), ht);
load至HBase表中。需要注意的点有：HBase表需要预创建region，不然在转化HFile时只有一个reduce（有几个region就有几个reduce，如何预创建分区见下一节），导致转化超慢；如此也会想到，在做HFile转化时，HBase表在做其他导入操作或者说有split操作，会导致后续的load也会超级慢，所以我们在做MapReduce HFile的转化时，HBase表最好不要有任何写入操作。
那这种方式能顺利进行吗？我们预创建表region数1000和100分别做了测试，在load数据时，都会报超时的异样（很抱歉，异常信息未记录），真是痛不欲生，花了很长时间没有解决此问题。
随即，我们又回到原导入方式，预先创建了region测试，速度上有点点提升，效率上本人着实觉得不合理，一个map运行很长时间。分析一下，map按行获取数据，转化成put对象后，执行导入，各种调优参数也调了批量导入也试过了，就是不见速度上有提升。为啥？很可能是在连接HBase时耗时，该MapRecude采用内置连接HBase的方式即：

conf.set(TableOutputFormat.OUTPUT_TABLE, hbase_table_name);
job.setOutputFormatClass(TableOutputFormat.class);
接着，本人不采用内置连接方式

protected void setup(Context context) throws IOException, InterruptedException {
    String table_name = context.getConfiguration().get("table_name").trim();
    if (StringUtils.isNotBlank(table_name)) {
        tablename = table_name;
        ht = new HTable(context.getConfiguration(), tablename);
        ht.setAutoFlush(false);
        ht.setWriteBufferSize(5*1024*1024);
        putlist = new ArrayList<Put>();
    }
}
......
protected void cleanup(Context context) throws IOException, InterruptedException {
    if (putlist.size() != 0){
        ht.put(putlist);
        putlist.clear();
    }
    ht.close();
}
立马map执行速度快了n倍，终于告了一段落。至于为何MapReduce内置连接HBase会慢，需查看源码，待研究。
常态化数据
由于常态化数据小，所以相应表采用普通的创建即可。HBase至HBase表操作采用org.apache.hadoop.hbase.mapreduce.CopyTable方式，HBase表至hive表采用MapReduce方式。
RowKey设计
使用HBase做存储，RowKey设计是最最关键的一部分。HBase查询数据方式有三种：

根据RowKey（高效）
根据RowKey区间（高效）
采用scan（转化成MapReduce低效，不满足实时性）
可见好的RowKey直接影响查询速率（此处未考虑secondary indexes）。

在数据导入过程中，需要预先创建region，也就是需要划分RowKey段。我们设计RowKey是数字型，对已有的初始化数据RowKey化分了1000份，经过测试，HBase会出现死节点情况，一个regionserver多少个region才合适呢？官方给出的答案是：

http://hbase.apache.org/book.html，2.5.2.6.1.how many regions per regions。

根据我们的实际情况，目前hadoop9个节点，4个regionserver。将1000份减少到100份，并根据数据量的大小，每个region设置成6G

public static boolean createTable(String tableName, String family, byte[][] splits) throws IOException {
    HBaseAdmin admin = new HBaseAdmin(conf);
    try {
        if (admin.tableExists(tableName)) {
            System.out.println("table already exists!");
        } else {
            HTableDescriptor tableDesc = new HTableDescriptor(tableName);
            tableDesc.addFamily(new HColumnDescriptor(family));
            tableDesc.setMaxFileSize(new Long("6442450900"));
            admin.createTable(tableDesc, splits);
            return true;
        }
        return false;
    } catch (TableExistsException e) {
        return false;
    }
}
public static byte[][] getHexSplits(String[] regions) {
    int numRegions = regions.length;
    byte[][] splits = new byte[numRegions - 1][];
    for (int i = 0; i < numRegions - 1; i++) {
        BigInteger key = new BigInteger(regions[i], 16);
        byte[] b = String.format("%016x", key).getBytes();
        splits[i] = b;
    }
    return splits;
}
应用总结
适合TB级别数据，将会增长到TB级别数据。
很高的写吞吐量，瞬间写入量大。
可通过rowkey访问数据，hbase rowkey访问数据最高效。
列可扩展。
结构化、半结构化数据。
无交叉表、连接表、多层索引、事务操作的数据模型。
遗留的问题
secondary indexesCCHADOOP-95 - hbase secondary indexes - OPEN ，0.94版本没有此功能，期待后续版本提供类似select * from table where anycell="XXX"的功能，目前国内有公司在应用层实现创建索引表，在应用层实现索引查询功能。
load HFile时超时的异样
表设计中，热数据表和增量冷数据表数据小，也采用了HBase存储。可通过关系型数据做中间存储替换，或者将3张表合并成一张大表替换（前提索引问题解决）。

评论

2 楼 lookqlp 2014-08-04  

di1984HIT 写道

谢谢，您写的很好啊，你的rowkey是基于什么创建的。？

根据不同的业务场景，rowkey方式都不一样，所以业务依赖性挺强的

1 楼 di1984HIT 2014-06-18  

谢谢，您写的很好啊，你的rowkey是基于什么创建的。？