很早就知道bulk load這個東西，也大致都清楚怎麼回事，居然直到前幾天才第一次實踐... 

這篇文章大致分為三個部分：

1. 使用hbase自帶的importtsv工具

2. 自己實現寫mr生成hfile並載入

3. bulk load本身及對依賴的第三方包的一些總結

第一部分：

匯入的檔案是data.txt，符合tsv格式，如下：

做一些準備工作：

a. 在hdfs上穿件/test目錄，並將data.txt傳至該目錄下

b. 建立hbase表bl_tmp

c. 將依賴的jar加到$HADOOP_HOME/conf/hadoop-env.sh (每個人的不一定一樣，加你需要的)

執行hbase自帶的imprttsv工具，這裡輸出路徑是output，列的定義由-Dimporttsv.columns指定：

程式正常執行，執行成功後，檢視/output目錄，output目錄下會根據列族名生成一個自錄，這裡是d，d目錄下為具體的hfile檔案：

執行completebulkload工具將hfile裝載到表bl_tmp中：

裝載完之後,d目錄下的hfile不存在了，這時查詢bl_tmp表，如下：

第二部分：

原始碼直接貼了，簡明扼要，沒什麼好說的... 關鍵的點詳見前邊兩篇簡要介紹相關原始碼的博文...

import java.io.IOException;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat;
import org.apache.hadoop.hbase.mapreduce.PutSortReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public final class HBaseBulkLoadDemo extends Configured implements Tool {

	public static class BulkLoadDemoMapper extends
			Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {

		private static final byte[] FAMILY_NAME = "d".getBytes();
		private static final byte[] COLUMN_A = "colA".getBytes();
		private static final byte[] COLUMN_B = "colB".getBytes();
		private static final byte[] COLUMN_C = "colC".getBytes();

		protected void map(LongWritable key, Text value, Context context) throws IOException,
				InterruptedException {
			String line = value.toString();
			String[] fields = line.split("\t");
			byte[] rowkeybytes = Bytes.toBytes(fields[0]);
			ImmutableBytesWritable rowkey = new ImmutableBytesWritable(rowkeybytes);
			Put put = new Put(rowkeybytes);
			put.add(FAMILY_NAME, COLUMN_A, fields[1].getBytes());
			put.add(FAMILY_NAME, COLUMN_B, fields[2].getBytes());
			put.add(FAMILY_NAME, COLUMN_C, fields[3].getBytes());
			context.write(rowkey, put);
		}

	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		System.exit(ToolRunner.run(new HBaseBulkLoadDemo(), args));
	}

	public int run(String[] args) throws Exception {
		Configuration conf = HBaseConfiguration.create();
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if (otherArgs.length != 3) {
			System.err.println("Usage: <tableName> <inputDir> <outputDir>");
			System.exit(2);
		}
		HTable table = new HTable(conf, otherArgs[0]);
		Job job = new Job(conf);
		job.setJarByClass(HBaseBulkLoadDemo.class);
		job.setJobName("HBaseBulkLoadDemo " + new Date());
		job.setMapperClass(BulkLoadDemoMapper.class);
		job.setReducerClass(PutSortReducer.class);
		job.setMapOutputKeyClass(ImmutableBytesWritable.class);
		job.setMapOutputValueClass(Put.class);
		FileInputFormat.addInputPath(job, new Path(otherArgs[1]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[2]));
		HFileOutputFormat.configureIncrementalLoad(job, table);
		return job.waitForCompletion(true) ? 0 : 1;
	}

}
 

程式用打包後，扔到叢集上執行，為驗證結果，注意先truncate掉bl_tmp表並刪掉/output目錄。

另外一點，這裡執行自己打的包，如果你沒有打依賴包的話，因為你用到hbase-version.jar，所以你需要把它加到HADOOP_CLASSPATH上：

執行自己打的jar包：

程式正確執行，檢視/output下的輸出：

將資料裝在進bl_tmp仍然可以用completebulkload工具，或者你可以自己寫一個工具，非常簡單，就是構造一個LoadIncrementalHFile物件，並呼叫它的doBulkLoad方法就好了。 然後檢視這時的bl_tmp(注意列名，與importtsv時不一樣...)：

第三部分：

關於bulkload本身：

1.這種方式適合初次匯入，對於大資料量，效率非常可觀，並且不需要表offline

2.目前貌似只適合每次對一個單列族匯入..

3.資料量很大時，因為reduce個數與region個數對應，所以導數前記得對錶進行預分割槽。

4.自己實現時，map階段的輸出只能是<ImmutableBytesWritable,KeyValue>或者<ImmutableBytesWritable,Put>，對應的reducer分別是KeyValueSortReducer和PutSortReducer。

關於hadoop對jar的載入方式及bulk load時第三方jar的一些說明，自己在實踐的時候起初迷惑了很久，所以特意總結了下：

1.hadoop jar在執行時一定會將HADOOP_CLASSPATH加到CLASSPATH上(感興趣可以cat hadoop看下)，並且將hadoop jar執行的目標jar拷貝到子節點。
2.依賴的第三方jar,一般三種方式處理，要麼-libjars，要麼加到HADOOP_HOME/lib下(所有子節點)，要麼打包進目標jar。
3.執行hadop jar hbase-version.jar importtsv時，由於將依賴的jar加到了HADOOP_CLASSPATH，並且在主節點本地可以找到，所以依託TableMapReduceUtil.addDependencyJars方法的作用，依賴的第三方jar在執行時被作為分散式快取拷貝到了子節點，程式得以正確執行。

完...