批量Load到HBase

hbase提供了写的操作，通常，我们可以采用HBase的Shell 客户端或者Java API进行操作。

如果数据量大的话，这两种操作是很费时的。其实如果了解了HBase的数据底层存储的细节的话，HBase的数据存储格式是HFile定义的格式。

批量导入HBase主要分两步：

• 通过mapreduce在输出目录OutputDir下生成一系列按Store存储结构一样的，存储HFile文件
• 通过LoadIncrementalHFiles.doBulkLoad把OutputDir里面的数据导入HBase表中

优点

1. HBase提供了一种直接写hfile文件的类，同时通过类似传统数据库的load把这些文件写进去，不再需要通过客户端或Java API一条一条插进去，
2. 这些接口简单方便，快捷灵活；
3. 应用不需要一直去连HBase集群进行RPC multi写，提高mapreduce效率；
4. HBase集群也相应减少不必要的连接，可以让它去多干些其它的事，效率更加高效，降低HBase集群因为大量并发写而产生不必要的风险。

1. 从HDFS批量导入

在MapReduce里面就把想要的输出成HFileOutputFormat格式的文件，然后通过LoadIncrementalHFiles.doBulkLoad方式就可以load进去即可。例子如下： 

Configuration conf = getConf();
conf.set("hbase.table.name", args[2]);
// Load hbase-site.xml
HBaseConfiguration.addHbaseResources(conf);
Job job = new Job(conf, "HBase Bulk Import Example");
job.setJarByClass(Mapper2.class);
job.setMapperClass(Mapper2.class);
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
job.setMapOutputValueClass(KeyValue.class);
job.setInputFormatClass(TextInputFormat.class);

// Auto configure partitioner and reducer
HTable hTable = new HTable(conf, args[2]);
HFileOutputFormat.configureIncrementalLoad(job, hTable);

FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.waitForCompletion(true);

// Load generated HFiles into table
LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
loader.doBulkLoad(new Path(args[1]), hTable);

2. 从MySQL批量导入

这个星期把一些MySQL表导到线上HBase表。这个MySQL表散了100份，在HBase集群未提供向业务使用时，通过Sqoop工具导进HBase表所花费的时间大约32个小时(已串行化)，在hbase集群繁忙时，花了10个小时都还没有把一张表导到HBase里面。这是有原因的，Sqoop未实现批量导的功能，它通常是边读边写。

后来自己写了一个从MySQL批量导入HBase的应用程序，每个表导入HBase所需时间平均只需要8分钟。

核心代码如下：

HBaseConfiguration.addHbaseResources(conf);

    Job job = new Job(conf, "Load_MySQL_" + table + "_to_HBase_" + hbaseTable);
    // 用来读mysql的Mapper
    job.setJarByClass(MysqlMapper.class);

    job.setMapperClass(MysqlMapper.class);
    job.setMapOutputKeyClass(ImmutableBytesWritable.class);
    job.setMapOutputValueClass(KeyValue.class);
    //配置DB参数
    DBConfiguration.configureDB(job.getConfiguration(), driver, connect, username, password);
    DataDrivenDBInputFormat.setInput(job, dbWritableClass, query, boundaryQuery);
    DataDrivenDBInputFormat.setInput(job, dbWritableClass, table, conditions, splitBy, columns);
    //设置输出路径
    FileOutputFormat.setOutputPath(job, new Path(tmpTargetDir));
    // 自动设置partitioner和reduce
    HTable hTable = new HTable(conf, hbaseTable);
    HFileOutputFormat.configureIncrementalLoad(job, hTable);

    job.waitForCompletion(true);

    // 上面JOB运行完后，就把数据批量load到HBASE中
    LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
    loader.doBulkLoad(new Path(tmpTargetDir), hTable);