hadoop执行hdfs文件到hbase表插入操作(xjl456852原创)

create ‘ns2:user‘, ‘info‘

1,xiejl,20
2,haha,30
3,liudehua,40
4,daoming,41

package com.xjl456852.mapreduce;
import org.apache.hadoop.fs.Path;
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.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import java.io.IOException;
/**
 * 将hdfs中的文本文件写入到hbase的表中
 * 程序的运行需要加入hadoop的配置文件和hbase的配置文件到jar包中
 * 对应的hbase的表
 * create ‘ns2:user‘,‘info‘
 * 
 * Created by xiejl on 2016/8/10.
 */
public class HBaseApp {
	public static void main(String [] args) {
		try {
			Job job = Job.getInstance();
			job.setJobName("text into hbase table");
			job.setJarByClass(HBaseApp.class);
			FileInputFormat.addInputPath(job, new Path(args[0]));
			//设置表名
			job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, args[1]);
			//设置输出格式为table
			job.setOutputFormatClass(TableOutputFormat.class);
			//设置输出的key类型为ImmutableBytesWritable
			job.setOutputKeyClass(ImmutableBytesWritable.class);
			//设置输出的value类型为Put
			job.setOutputValueClass(Put.class);
			//因为map输出key和reduce输出的key类型不一致,所以需要再设置map的key输出类型为Text
			job.setMapOutputKeyClass(Text.class);
			//因为map输出value和reduce输出的value类型不一致,所以需要再设置map的value输出类型为Text
			job.setMapOutputValueClass(Text.class);
			//Mapper
			job.setMapperClass(MyMapper.class);
			//Reducer
			job.setReducerClass(MyReducer.class);
			System.exit(job.waitForCompletion(true) ? 0 : 1);
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
	}
}

package com.xjl456852.mapreduce;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
/**
 * Created by xiejl on 2016/8/10.
 */
public class MyMapper extends Mapper<LongWritable, Text, Text, Text> {
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		String line = value.toString();
		int index = line.indexOf(",");
		String rowKey = line.substring(0, index);
		//跳过逗号
		String valueLine = line.substring(index+1);
		context.write(new Text(rowKey), new Text(valueLine));
		
	}
}

package com.xjl456852.mapreduce;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
/**
 * Created by xiejl on 2016/8/11.
 */
public class MyReducer extends Reducer<Text, Text, ImmutableBytesWritable, Put> {
	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		byte[] rowKey = Bytes.toBytes(key.toString());
		for(Text text : values) {
			//设置put对象的行键
			Put put = new Put(rowKey);
			String line = text.toString();
			int index = line.indexOf(",");
			String name = line.substring(0, index);
			String age = line.substring(index+1);
			//列族的是建表时固定的,列和值是插入时添加的.
			put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"),Bytes.toBytes(name));
			put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"),Bytes.toBytes(age));
			
			context.write(new ImmutableBytesWritable(Bytes.toBytes(key.toString())), put);
		}
	}
}

TEMP=`ls /opt/modules/hbase/lib/*.jar`
HBASE_JARS=`echo $TEMP | sed ‘s/ /:/g‘`
HADOOP_CLASSPATH=$HBASE_JARS

hadoop jar hbase.jar com.xjl456852.mapreduce.HBaseApp data/hbase_input ns2:user