从hadoop取出文件写入hbase表中

yesn

package example2;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
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.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
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.util.GenericOptionsParser;
//Administrator
public class ImportFromFileExample {
public static class ImportMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable,Text >{

@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
context.write(new ImmutableBytesWritable(Bytes.toBytes(key.get())), value);
}

}
public static class Reducer1 extends TableReducer<ImmutableBytesWritable, Text, ImmutableBytesWritable> {
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];
}
}
public void reduce(ImmutableBytesWritable key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
String valueCon=null;
for(Text text:values){
valueCon+=text.toString();
}
Put put = new Put(key.get());
put.add(family, qualifier, Bytes.toBytes(valueCon));
context.write(key, put);
}
}
/**
* @param args
*/
public static void main(String[] args)throws Exception {
Configuration conf=HBaseConfiguration.create();
String []argArray=new GenericOptionsParser(conf, args).getRemainingArgs();
if(argArray.length!=1){
System.exit(1);
}
conf.set("conf.column", "family1:text");
Job job=new Job(conf,"import from hdfs to hbase");
job.setJarByClass(ImportFromFileExample.class);
job.setMapperClass(ImportMapper.class);
job.setOutputFormatClass(TableOutputFormat.class);
job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, "testtable");
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(ImmutableBytesWritable.class);
job.setOutputValueClass(Put.class);
TableMapReduceUtil.initTableReducerJob("testtable", Reducer1.class, job);
FileInputFormat.addInputPaths(job, argArray[0]);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}