Hbase 集成 MapReduce（三）

package com.hbase.test02_with_mapreduce;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
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.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * User2BasicMapReduce
 * 将user表中info:name,info:age两列复制到basic表中(basic表事先必须存在！)
 * @author zhangqingli
 *
 */
public class User2BasicMapReduce extends Configured implements Tool {
	/**
	 * Mapper
	 */
	public static class ReadUserMapper extends TableMapper<Text, Put> {
		private Text mapOutputKey = new Text();

		@Override
		public void map(ImmutableBytesWritable key, Result value,
				Mapper<ImmutableBytesWritable, Result, Text, Put>.Context context)
				throws IOException, InterruptedException {
			// get rowkey
			String rowKey = Bytes.toString(key.get());
			// set
			mapOutputKey.set(rowKey);

			Put put = new Put(key.get());
			for (Cell cell : value.rawCells()) {
				// add family:qualifier
				if ("info".equals(Bytes.toString(CellUtil.cloneFamily(cell)))) {
					// add column
					if ("name".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))) {
						put.add(cell);
					}
					if ("age".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))) {
						put.add(cell);
					}
				}
			}
			//
			context.write(mapOutputKey, put);
		}
	}

	/**
	 * Reducer
	 */
	public static class WriteBasicReducer 
      		extends TableReducer<Text, Put, ImmutableBytesWritable> {
		@Override
		protected void reduce(Text key, Iterable<Put> values,
				Reducer<Text, Put, ImmutableBytesWritable, Mutation>.Context context)
				throws IOException, InterruptedException {
			for (Put put : values) {
				context.write(null, put);
			}
		}
	}

	/**
	 * Driver
	 */
	@Override
	public int run(String[] args) throws Exception {
		// create job
		Job job = Job.getInstance(this.getConf(), this.getClass().getSimpleName());
		
		// set runnable job class
		job.setJarByClass(this.getClass());
		// set job
		Scan scan = new Scan();
		scan.setCaching(500); 		// default is 1, which will be bad for MapReduce jobs
		scan.setCacheBlocks(false); // don't set to true for MR jobs
		// set input & mapper
		TableMapReduceUtil.initTableMapperJob(
          		"user", // input table
				scan,   // Scan instance to control CF and attribute selection
				ReadUserMapper.class, 	// mapper class
				Text.class, 			// mapper output key
				Put.class, 				// mapper output value
				job);
		// set reducer & output
		TableMapReduceUtil.initTableReducerJob("basic", WriteBasicReducer.class, job);
		job.setNumReduceTasks(1); 
		
		boolean isSuccess = job.waitForCompletion(true);
		return isSuccess ? 0 : 1;
	}

	/**
	 * main
	 */
	public static void main(String[] args) throws Exception {
		// get conf
		Configuration conf = HBaseConfiguration.create();
		// submit job
		int status = ToolRunner.run(conf, new User2BasicMapReduce(), args);
		// exit program
		System.exit(status);
	}
}