Jaki jest najszybszy sposób programowego ładowania danych do bazy HBase?

Mam zwykły plik tekstowy z prawdopodobnie milionami linii, który wymaga niestandardowego parsowania i chcę go jak najszybciej załadować do tabeli HBase (używając Hadoop lub HBase Java client).

Moje obecne rozwiązanie opiera się na pracy MapReduce BEZ części Reduce. Używam FileInputFormat do odczytu pliku tekstowego, aby każda linia była przekazywana do metody map mojej klasy Mapper. W tym momencie linia jest przetwarzana do postaci Put obiektu, który jest zapisywany do context. Następnie, TableOutputFormat bierze Put obiekt i wstawia go do tabeli.

To rozwiązanie daje średnią szybkość wstawiania 1000 wierszy na sekundę, czyli mniej niż się spodziewałem. moja konfiguracja HBase jest w trybie pseudo rozproszonym na jednym serwerze.

Ciekawostką jest to, że podczas wstawiania 1,000,000 wierszy, 25 maperów (zadań) jest wywoływanych, ale działają seryjnie (jeden po drugim); czy to normalne?

Oto kod do mojego obecnego rozwiązania:

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);
}