title: What is the fastest way to bulk load data into HBase programmatically?
body: <p>I have a Plain text file with possibly millions of lines which needs custom parsing and I want to load it into an HBase table as fast as possible (using Hadoop or HBase Java client).</p>
  
  <p>My current solution is based on a <strong>MapReduce</strong> job without the Reduce part. I use <code>FileInputFormat</code> to read the text file so that each line is passed to the <code>map</code> method of my <code>Mapper</code> class. At this point the line is parsed to form a <code>Put</code> object which is written to the <code>context</code>. Then, <code>TableOutputFormat</code> takes the <code>Put</code> object and inserts it to table.</p>
  
  <p>This solution yields an average insertion rate of 1,000 rows per second, which is less than what I expected. <strong>My HBase setup is in pseudo distributed mode on a single server.</strong></p>
  
  <p>One interesting thing is that during insertion of 1,000,000 rows, 25 Mappers (tasks) are spawned but they run serially (one after another); is this normal?</p>
  
  <p>Here is the code for my current solution:</p>
  
  <pre><code>public static class CustomMap extends Mapper&lt;LongWritable, Text, ImmutableBytesWritable, Put&gt; {
  
      protected void map(LongWritable key, Text value, Context context) throws IOException {
          Map&lt;String, String&gt; 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);
  }
  </code></pre>