MapReduce从HDFS读取数据存储到HBase中
现有HDFS中有一个student.txt文件，格式如下
95002,刘晨,女,19,IS 95017,王风娟,女,18,IS 95018,王一,女,19,IS 95013,冯伟,男,21,CS 95014,王小丽,女,19,CS 95019,邢小丽,女,19,IS 95020,赵钱,男,21,IS 95003,王敏,女,22,MA 95004,张立,男,19,IS 95012,孙花,女,20,CS 95010,孔小涛,男,19,CS 95005,刘刚,男,18,MA 95006,孙庆,男,23,CS 95007,易思玲,女,19,MA 95008,李娜,女,18,CS 95021,周二,男,17,MA 95022,郑明,男,20,MA 95001,李勇,男,20,CS 95011,包小柏,男,18,MA 95009,梦圆圆,女,18,MA 95015,王君,男,18,MA
将HDFS上的这个文件里面的数据写入到HBase数据块中
MapReduce实现代码如下
import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil; import org.apache.hadoop.hbase.mapreduce.TableReducer; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; 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.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; public class ReadHDFSDataToHbaseMR extends Configured implements Tool{ public static void main(String[] args) throws Exception { int run = ToolRunner.run(new ReadHDFSDataToHbaseMR(), args); System.exit(run); } @Override public int run(String[] arg0) throws Exception { Configuration conf = HBaseConfiguration.create(); conf.set("fs.defaultFS", "hdfs://myha01/"); conf.set("hbase.zookeeper.quorum", "hadoop1:2181,hadoop2:2181,hadoop3:2181"); System.setProperty("HADOOP_USER_NAME", "hadoop"); FileSystem fs = FileSystem.get(conf); // conf.addResource("config/core-site.xml"); // conf.addResource("config/hdfs-site.xml"); Job job = Job.getInstance(conf); job.setJarByClass(ReadHDFSDataToHbaseMR.class); job.setMapperClass(HDFSToHbaseMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(NullWritable.class); TableMapReduceUtil.initTableReducerJob("student", HDFSToHbaseReducer.class, job,null,null,null,null,false); job.setOutputKeyClass(NullWritable.class); job.setOutputValueClass(Put.class); Path inputPath = new Path("/student/input/"); Path outputPath = new Path("/student/output/"); if(fs.exists(outputPath)) { fs.delete(outputPath,true); } FileInputFormat.addInputPath(job, inputPath); FileOutputFormat.setOutputPath(job, outputPath); boolean isDone = job.waitForCompletion(true); return isDone ? 0 : 1; } public static class HDFSToHbaseMapper extends Mapper<LongWritable, Text, Text, NullWritable>{ @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { context.write(value, NullWritable.get()); } } /** * 95015,王君,男,18,MA * */ public static class HDFSToHbaseReducer extends TableReducer<Text, NullWritable, NullWritable>{ @Override protected void reduce(Text key, Iterable<NullWritable> values,Context context) throws IOException, InterruptedException { String[] split = key.toString().split(","); Put put = new Put(split[0].getBytes()); put.addColumn("info".getBytes(), "name".getBytes(), split[1].getBytes()); put.addColumn("info".getBytes(), "sex".getBytes(), split[2].getBytes()); put.addColumn("info".getBytes(), "age".getBytes(), split[3].getBytes()); put.addColumn("info".getBytes(), "department".getBytes(), split[4].getBytes()); context.write(NullWritable.get(), put); } } }
MapReduce从HBase读取数据计算平均年龄并存储到HDFS中
import java.io.IOException; import java.util.List; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellUtil; import org.apache.hadoop.hbase.HBaseConfiguration; 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.util.Bytes; import org.apache.hadoop.io.DoubleWritable; import org.apache.hadoop.io.IntWritable; 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.util.Tool; import org.apache.hadoop.util.ToolRunner; public class ReadHbaseDataToHDFS extends Configured implements Tool{ public static void main(String[] args) throws Exception { int run = ToolRunner.run(new ReadHbaseDataToHDFS(), args); System.exit(run); } @Override public int run(String[] arg0) throws Exception { Configuration conf = HBaseConfiguration.create(); conf.set("fs.defaultFS", "hdfs://myha01/"); conf.set("hbase.zookeeper.quorum", "hadoop1:2181,hadoop2:2181,hadoop3:2181"); System.setProperty("HADOOP_USER_NAME", "hadoop"); FileSystem fs = FileSystem.get(conf); // conf.addResource("config/core-site.xml"); // conf.addResource("config/hdfs-site.xml"); Job job = Job.getInstance(conf); job.setJarByClass(ReadHbaseDataToHDFS.class); // 取对业务有用的数据 info,age Scan scan = new Scan(); scan.addColumn("info".getBytes(), "age".getBytes()); TableMapReduceUtil.initTableMapperJob( "student".getBytes(), // 指定表名 scan, // 指定扫描数据的条件 HbaseToHDFSMapper.class, // 指定mapper class Text.class, // outputKeyClass mapper阶段的输出的key的类型 IntWritable.class, // outputValueClass mapper阶段的输出的value的类型 job, // job对象 false ); job.setReducerClass(HbaseToHDFSReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(DoubleWritable.class); Path outputPath = new Path("/student/avg/"); if(fs.exists(outputPath)) { fs.delete(outputPath,true); } FileOutputFormat.setOutputPath(job, outputPath); boolean isDone = job.waitForCompletion(true); return isDone ? 0 : 1; } public static class HbaseToHDFSMapper extends TableMapper<Text, IntWritable>{ Text outKey = new Text("age"); IntWritable outValue = new IntWritable(); // key是hbase中的行键 // value是hbase中的所行键的所有数据 @Override protected void map(ImmutableBytesWritable key, Result value,Context context) throws IOException, InterruptedException { boolean isContainsColumn = value.containsColumn("info".getBytes(), "age".getBytes()); if(isContainsColumn) { List<Cell> listCells = value.getColumnCells("info".getBytes(), "age".getBytes()); System.out.println("listCells: "+listCells); Cell cell = listCells.get(0); System.out.println("cells: "+cell); byte[] cloneValue = CellUtil.cloneValue(cell); String ageValue = Bytes.toString(cloneValue); outValue.set(Integer.parseInt(ageValue)); context.write(outKey,outValue); } } } public static class HbaseToHDFSReducer extends Reducer<Text, IntWritable, Text, DoubleWritable>{ DoubleWritable outValue = new DoubleWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException { int count = 0; int sum = 0; for(IntWritable value : values) { count++; sum += value.get(); } double avgAge = sum * 1.0 / count; outValue.set(avgAge); context.write(key, outValue); } } }