Hadoop Mapper 阶段将数据直接从 HDFS 导入 Hbase

20130512	1	-1	-1	13802	1	2013-05-12 07:26:22	
20130512	1	-1	-1	13802	1	2013-05-12 11:18:24

package WebsiteAnalysis;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class Map2Hdfs {
	public static final String NAME = "ImportFromFile";

	public enum Counters {
		LINES
	}

	static class ImportMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Writable> {
		private byte[] family = null;
		private byte[] qualifier = null;

		@Override
		protected void setup(Context context) throws IOException, InterruptedException {
			String column = context.getConfiguration().get("conf.column");
			byte[][] colkey = KeyValue.parseColumn(Bytes.toBytes(column));
			family = colkey[0];
			if (colkey.length > 1) {
				qualifier = colkey[1];
			}
		}

		@Override
		public void map(LongWritable offset, Text line, Context context) throws IOException {
			try {
				String[] lineArr = line.toString().split("\t");
				Put put = new Put(Bytes.toBytes(offset + ""));
				put.add(family, Bytes.toBytes("time"), Bytes.toBytes(lineArr[lineArr.length - 1]));
				context.write(new ImmutableBytesWritable(Bytes.toBytes(offset + "")), put);
				context.getCounter(Counters.LINES).increment(1);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = HBaseConfiguration.create();
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		conf.set("conf.column", "cf");
		String inputPath = "/dsap/middata/lj/ooxx/pv";
		Job job = new Job(conf, "TestMap2Hdfs");

		job.setJarByClass(Map2Hdfs.class);
		job.setMapperClass(ImportMapper.class);
		job.setOutputFormatClass(TableOutputFormat.class);
		job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, "TestMap2Hdfs");
		job.setOutputKeyClass(ImmutableBytesWritable.class);
		job.setOutputValueClass(Writable.class);
		job.setNumReduceTasks(0);
		FileInputFormat.addInputPath(job, new Path(inputPath + "/" + otherArgs[0]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}