= 範例七： LoadHBaseMapper = 

 == 說明：  == 
	此程式碼將HBase的資料取出來，再將結果塞回hdfs上

 == 執行： ==

[raw-attachment:wiki:NCHCCloudCourse100204:tsmcHBase_100204.jar 測試用打包檔 tsmcHBase_100204.jar]

{{{
	bin/hadoop jar tsmcHBase_100204.jar LoadHBaseMapper
}}}
 == 結果： == 
{{{
 $ hadoop fs -cat <hdfs_output>/part-r-00000

	54 30 31	GunLong
	54 30 32	Esing
	54 30 33	SunDon
	54 30 34	StarBucks
}}}
== 注意： ==
1.	請注意hbase 上必須要有 table, 並且已經有資料

2.	運算完後，程式將執行結果放在你指定 hdfs的<hdfs_output> 內
 		請注意 沒有 <hdfs_output> 資料夾


{{{
#!java
package tsmc;

// LoadHBaseMapper
//說明： 
//	此程式碼將HBase的資料取出來，再將結果塞回hdfs上
//
//運算方法：
//	將此程式運作在hadoop 0.20 平台上，用(參考2)的方法加入hbase參數後，將此程式碼打包成XX.jar 
//  執行：
//	---------------------------
//	hadoop jar XX.jar LoadHBaseMapper <hdfs_output>
//	---------------------------
//
//結果：
// $ hadoop fs -cat <hdfs_output>/part-r-00000
// ---------------------------
//	54 30 31	GunLong
//	54 30 32	Esing
//	54 30 33	SunDon
//	54 30 34	StarBucks
// ---------------------------
//注意：
//1.	請注意hbase 上必須要有 table, 並且已經有資料
//2.	運算完後，程式將執行結果放在你指定 hdfs的<hdfs_output> 內
// 		請注意 沒有 <hdfs_output> 資料夾

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
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.TableInputFormat;
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.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class LoadHBaseMapper {
	public static class HtMap extends TableMapper<Text, Text> {

		public void map(ImmutableBytesWritable key, Result value,
				Context context) throws IOException, InterruptedException {

			String res = Bytes.toString(value.getValue(Bytes.toBytes("Detail"),
					Bytes.toBytes("Name")));

			context.write(new Text(key.toString()), new Text(res));

		}
	}

	// TableReducer<KEYIN,VALUEIN,KEYOUT>
	// 原本為 TableReducer<Text, IntWritable, NullWritable >
	// 但在此改成 LongWritable 也可以
	// 因此證明在此的Class可以很多,org.apache.hadoop.io.* 內有write()的Writable class應該皆可
	public static class HtReduce extends Reducer<Text, Text, Text, Text> {

		public void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {

			String str = new String("");
			Text final_key = new Text(key);
			Text final_value = new Text();
			// 將key值相同的values，透過 && 符號分隔之
			for (Text tmp : values) {
				str += tmp.toString();
			}
			final_value.set(str);
			context.write(final_key, final_value);
		}
	}

	public static void main(String args[]) throws Exception {
		// debug
		String[] argv = { "output-lhm" };
		args = argv;
		String input = args[0];

		String tablename = "tsmc";

		Configuration conf = new Configuration();

		Job job = new Job(conf, tablename + " hbase data to hdfs");
		job.setJarByClass(LoadHBaseMapper.class);

		// // 設定哪張hbase的table為輸入
		conf.set(TableInputFormat.INPUT_TABLE, tablename);
		Scan myScan = new Scan("".getBytes(), "12".getBytes());
		myScan.addColumn("Detail:Name".getBytes());

		// 先用 TableMapReduceUtil 的 initTableMapperJob 與 initTableReducerJob 來做設定
		TableMapReduceUtil.initTableMapperJob(tablename, myScan, HtMap.class,
				Text.class, Text.class, job);

		job.setMapperClass(HtMap.class);
		job.setReducerClass(HtReduce.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		job.setInputFormatClass(TableInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		FileOutputFormat.setOutputPath(job, new Path(input));

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