HBase - TableInputFormat

這篇主要記錄如何將HBase(0.20.3)當作Hadoop MapReduce程式的輸入來源，下述的程式碼很單純的從一個Table取出資料並直接輸出至HDFS上：

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class TestTableInput
{
	public static class Map extends TableMapper<Text, Text>
	{
		@Override
		public void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException
		{
			String v = Bytes.toString(value.value());
			context.write(new Text(Bytes.toString(value.getRow())), new Text(v));
		}
	}

	public static class Reduce extends Reducer<Text, Text, Text, Text>
	{
		@Override
		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException
		{
			context.write(key, new Text(values.iterator().next()));
		}
	}

	public static void main(String args[]) throws Exception
	{
		if(args.length != 2)
		{
			System.out.println("Usage: hadoop jar TestTableInput.jar <table> <output>");
			System.exit(-1);
		}
		
		String tablename = args[0];
		String output = args[1];

		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(conf);
		fs.delete(new Path(output), true);

		Job job = new Job(conf, "TestTableInput");
		Scan scan = new Scan();
		TableMapReduceUtil.initTableMapperJob(tablename, scan, Map.class, Text.class, Text.class, job);	
		job.setJarByClass(TestTableInput.class);
		job.setReducerClass(Reduce.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		TextOutputFormat.setOutputPath(job, new Path(output));

		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}

這裡值得注意的有二個地方，一個是Map必須繼承HBase所提供的TableMapper(其實TableMapper也是繼承於Mapper)，它用來指定KEYIN和VALUEIN兩個類別，它們分別為：ImmutableBytesWritable、Result，而這兩個類別分別對應的Table資料如下：

．ImmutableBytesWritable = Row Key

．Result = Row Key+Column+Timestamp+Value

另一個值得注意的是在main方式中用到的TableMapReduceUtil.initTableMapperJob()方法，它封裝了一些設定，如下圖所示：

從圖中我們可以知道該方法會幫我們設定InputFormatClass為TableInputFormat，還有一些相關設定，例如：TableInputFormat.INPUT_TABLE用來設定輸入的Table，TableInputFormat.SCAN用來設定Scan(由於Scan是一個物件，所以必須透過convertScanToString()方法來轉碼成Base64編碼)