以编程方式将数据批量加载到HBase中的最快方法是什么

Question

我有一个可能有数百万行的纯文本文件，需要自定义解析，我想尽快将其加载到HBase表中（使用Hadoop或HBase Java客户端）。

我当前的解决方案基于MapReduce作业，不包含Reduce部分。我使用FileInputFormat读取文本文件，这样每一行都会传递给我的Mapper类的map方法。在这一点上，该行被解析以形成Put对象，该对象被写入context。然后，TableOutputFormat获取Put对象并将其插入到表中。

这个解决方案的平均插入速率为每秒1000行，低于我的预期我的HBase设置在单个服务器上处于伪分布式模式

一件有趣的事情是，在插入1000000行的过程中，生成了25个Mapper（任务），但它们是串行运行的（一个接一个）；这正常吗？

这是我当前解决方案的代码：

public static class CustomMap extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
    protected void map(LongWritable key, Text value, Context context) throws IOException {
        Map<String, String> parsedLine = parseLine(value.toString());
        Put row = new Put(Bytes.toBytes(parsedLine.get(keys[1])));
        for (String currentKey : parsedLine.keySet()) {
            row.add(Bytes.toBytes(currentKey),Bytes.toBytes(currentKey),Bytes.toBytes(parsedLine.get(currentKey)));
        }
        try {
            context.write(new ImmutableBytesWritable(Bytes.toBytes(parsedLine.get(keys[1]))), row);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}
public int run(String[] args) throws Exception {
    if (args.length != 2) {
        return -1;
    }
    conf.set("hbase.mapred.outputtable", args[1]);
    // I got these conf parameters from a presentation about Bulk Load
    conf.set("hbase.hstore.blockingStoreFiles", "25");
    conf.set("hbase.hregion.memstore.block.multiplier", "8");
    conf.set("hbase.regionserver.handler.count", "30");
    conf.set("hbase.regions.percheckin", "30");
    conf.set("hbase.regionserver.globalMemcache.upperLimit", "0.3");
    conf.set("hbase.regionserver.globalMemcache.lowerLimit", "0.15");
    Job job = new Job(conf);
    job.setJarByClass(BulkLoadMapReduce.class);
    job.setJobName(NAME);
    TextInputFormat.setInputPaths(job, new Path(args[0]));
    job.setInputFormatClass(TextInputFormat.class);
    job.setMapperClass(CustomMap.class);
    job.setOutputKeyClass(ImmutableBytesWritable.class);
    job.setOutputValueClass(Put.class);
    job.setNumReduceTasks(0);
    job.setOutputFormatClass(TableOutputFormat.class);
    job.waitForCompletion(true);
    return 0;
}
public static void main(String[] args) throws Exception {
    Long startTime = Calendar.getInstance().getTimeInMillis();
    System.out.println("Start time : " + startTime);
    int errCode = ToolRunner.run(HBaseConfiguration.create(), new BulkLoadMapReduce(), args);
    Long endTime = Calendar.getInstance().getTimeInMillis();
    System.out.println("End time : " + endTime);
    System.out.println("Duration milliseconds: " + (endTime-startTime));
    System.exit(errCode);
}

Answer 1

我经历了一个可能与您的过程非常相似的过程，试图找到一种将数据从MR加载到HBase的有效方法。我发现使用HFileOutputFormat作为MR 的OutputFormatClass是可行的

下面是我的代码的基础，我必须生成job和Mapper map函数来写入数据。这很快。我们不再使用它了，所以我手头没有数字，但不到一分钟就有250万张唱片。

这是我为MapReduce过程生成作业而编写的（精简的）函数，用于将数据放入HBase

private Job createCubeJob(...) {
    //Build and Configure Job
    Job job = new Job(conf);
    job.setJobName(jobName);
    job.setMapOutputKeyClass(ImmutableBytesWritable.class);
    job.setMapOutputValueClass(Put.class);
    job.setMapperClass(HiveToHBaseMapper.class);//Custom Mapper
    job.setJarByClass(CubeBuilderDriver.class);
    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(HFileOutputFormat.class);
    TextInputFormat.setInputPaths(job, hiveOutputDir);
    HFileOutputFormat.setOutputPath(job, cubeOutputPath);
    Configuration hConf = HBaseConfiguration.create(conf);
    hConf.set("hbase.zookeeper.quorum", hbaseZookeeperQuorum);
    hConf.set("hbase.zookeeper.property.clientPort", hbaseZookeeperClientPort);
    HTable hTable = new HTable(hConf, tableName);
    HFileOutputFormat.configureIncrementalLoad(job, hTable);
    return job;
}

这是我在HiveToHBaseMapper类中的map函数（经过轻微编辑）。

public void map(WritableComparable key, Writable val, Context context)
        throws IOException, InterruptedException {
    try{
        Configuration config = context.getConfiguration();
        String[] strs = val.toString().split(Constants.HIVE_RECORD_COLUMN_SEPARATOR);
        String family = config.get(Constants.CUBEBUILDER_CONFIGURATION_FAMILY);
        String column = strs[COLUMN_INDEX];
        String Value = strs[VALUE_INDEX];
        String sKey = generateKey(strs, config);
        byte[] bKey = Bytes.toBytes(sKey);
        Put put = new Put(bKey);
        put.add(Bytes.toBytes(family), Bytes.toBytes(column), (value <= 0) 
                        ? Bytes.toBytes(Double.MIN_VALUE)
                        : Bytes.toBytes(value));
        ImmutableBytesWritable ibKey = new ImmutableBytesWritable(bKey);
        context.write(ibKey, put);
        context.getCounter(CubeBuilderContextCounters.CompletedMapExecutions).increment(1);
    }
    catch(Exception e){
        context.getCounter(CubeBuilderContextCounters.FailedMapExecutions).increment(1);    
    }
}

我很确定这不会是一个副本&为您粘贴解决方案。显然，我在这里处理的数据不需要任何自定义处理（这是在本次之前的MR工作中完成的）。我想从中提供的主要内容是HFileOutputFormat。剩下的只是我如何使用它的一个例子。：）
我希望它能让你走上一条通往良好解决方案的坚实道路

Answer 2

一件有趣的事情是，在插入1000000行的过程中，生成了25个Mapper（任务），但它们是串行运行的（一个接一个）；这正常吗？

默认为2的mapreduce.tasktracker.map.tasks.maximum参数确定了节点上可以并行运行的最大任务数。除非更改，否则您应该看到每个节点上同时运行2个映射任务。