五、当HBase遇上MapReduce第1关HBase的MapReduce快速入门任务描述本关任务编写一个HBase的MapReduce配置程序。相关知识为了完成本关任务你需要掌握快速入门HBase的MapReduce配置HBase的Map类 和Reduce类快速入门要运行HBase的MapReduce很简单HBase内部已经自带了MapReduce任务。首先登录到服务器上然后确保当前的用户下有HADOOP_HOME和HBASE_HOME 这两个环境变量并启动Hadoop、HBase。默认情况下部署到MapReduce集群的MapReduce作业是不能访问HBase的配置和类的要为MapReduce作业提供他们所需的访问权限推荐的方法是使用 HADOOP_CLASSPATH让HBase添加它的依赖jar , HADOOP_CLASSPATH${HBASE_HOME}/bin/hbase classpath。我们先来看看hbase-mapreduce-版本号.jar包中包含了哪些MapReduce Job# HADOOP_CLASSPATH${HBASE_HOME}/bin/hbase classpath ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-mapreduce-2.0.0.jarAn example program must be given as the first argument.Valid program names are:CellCounter: Count cells in HBase table.WALPlayer: Replay WAL files.completebulkload: Complete a bulk data load.copytable: Export a table from local cluster to peer cluster.export: Write table data to HDFS.exportsnapshot: Export the specific snapshot to a given FileSystem.import: Import data written by Export.importtsv: Import data in TSV format.rowcounter: Count rows in HBase table.verifyrep: Compare data from tables in two different clusters. It doesnt work for incrementColumnValuesd cells since timestamp is changed after appending to WAL.可以看到里面有个叫rowcounter,我们这次就使用它了rowcounter做的事情很简单就是统计当前表有多少行当你调用hbase-mapreduce-版本号.jar并使用rowcounter为第一个参数的时候就会使用这个MapReduce 的job第二个参数就是你想统计的目标表所以这条命令的格式为$ HADOOP_CLASSPATH${HBASE_HOME}/bin/hbase classpath ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/hbase-mapreduce-版本号.jar rowcounter 目标表想要统计的表名为CAR_INFOS所以我们在这个例子中使用的命令为HADOOP_CLASSPATH${HBASE_HOME}/bin/hbase classpath ${HADOOP_HOME}/bin/hadoop jar ${HBASE_HOME}/lib/hbase-mapreduce-2.0.0.jar rowcounter CAR_INFOS运行后会输出很多日志信息在信息的末尾我们会看到这样一句话:org.apache.hadoop.hbase.mapreduce.RowCounter$RowCounterMapper$CountersROWS1004File Input Format CountersBytes Read0File Output Format CountersBytes Written0这个ROWS1004就是结果意思是mytable统计出来的行数是1004。HBase 的 MapReduce 配置从运行HBase的MapReduce命令可以看出HBase的MapReduce仍然使用的是Hadoop提供的原生MapReduce,使用hadoop命令来启动jar包。那HBase的MapReduce和原生的具体有何区别呢我们首先来看看HBase的MapReduce的配置信息。Configuration conf HBaseConfiguration.create(getConf());String tablename args[0];String targetTable args[1];Job job Job.getInstance(conf);Scan scan new Scan();TableMapReduceUtil.initTableMapperJob(tablename,scan,MyMapper.class, Text.class, IntWritable.class,job);TableMapReduceUtil.initTableReducerJob(targetTable,MyReducer.class,job);}可以看出原本繁琐配置不见了Map节点使用initTableMapJob 方法进行配置方法里依次传入输出表、HBase过滤器、mapper类、mapper输出key、mapper输入value、job类。Reducer节点的相关配置则通过initTableReducerJob方法进行配置方法依次传入 输出表、reducer类、job类就此HBase的MapReducer配置就完成了。HBase 的 Map 类 和 Reduce 类使用HBase的MapReducer则不能使用原生的Map类和Reducer类, HBase定义了继承Map类的TableMapper和 继承Reducer类的 TableReducer。如下public abst\fract class TableMapperKEYOUT, VALUEOUTextends MapperImmutableBytesWritable, Result, KEYOUT, VALUEOUT {}public abst\fract class TableReducerKEYIN, VALUEIN, KEYOUTextends ReducerKEYIN, VALUEIN, KEYOUT, Mutation {}对于Mapper类我们使用TableMapper时不再需要设定四个泛型类型只需要确定输出到Reducer的key和Value的类型而起始偏移量、及文本内容变为HBase特有的rowkey和行Result。对于Reduce类我们使用TableReducer时 最后写入的Value的类型不再自己设定而变为HBase特有的Mutation由Mutation存入最后的数据。编程要求请仔细阅读右侧代码根据方法内的提示在Begin - End区域内进行代码补充在run()方法增加MapReduce配置信息。String[] args数组第一个参数为输入表 第二个参数为输出表。测试说明补充完代码后点击测评平台会对你编写的代码进行测试当你的结果与预期输出一致时即为通过。测试输入t_comment t_word_count预期输出word:Iword_info:count 9word:beword_info:count 4word:isword_info:count 4word:theword_info:count 4word:toword_info:count 8word:withword_info:count 4word:youword_info:count 7开始你的任务吧祝你成功package com.processdata;import java.io.IOException;import java.util.List;import java.util.Scanner;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.conf.Configured;import org.apache.hadoop.hbase.HBaseConfiguration;import org.apache.hadoop.hbase.client.Connection;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.IntWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.util.Tool;import org.apache.hadoop.util.ToolRunner;import org.apdplat.word.WordSegmenter;import org.apdplat.word.segmentation.Word;import com.util.HBaseUtil;import com.vdurmont.emoji.EmojiParser;/*** 词频统计**/public class WorldCountMapReduce extends Configured implements Tool {private static class MyMapper extends TableMapperText, IntWritable {private static byte[] family comment_info.getBytes();private static byte[] column content.getBytes();Overrideprotected void map(ImmutableBytesWritable rowKey, Result result, Context context) {try {byte[] value result.getValue(family, column);String content new String(value, utf-8);String[] split content.split( );for (String str : split) {Text text new Text(str);IntWritable v new IntWritable(1);context.write(text, v);}} catch (Exception e) {e.printStackTrace();}}}private static class MyReducer extends TableReducerText, IntWritable, ImmutableBytesWritable {private static byte[] family word_info.getBytes();private static byte[] column count.getBytes();Overridepublic void reduce(Text key, IterableIntWritable values, Context context) {int sum 0;for (IntWritable value : values) {sum value.get();}Put put new Put(Bytes.toBytes(key.toString()));put.addColumn(family, column, Bytes.toBytes(sum));try {context.write(null, put);} catch (IOException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();}}}Overridepublic int run(String[] args) throws Exception {// 配置Job/********** Begin *********/// 创建Conf对象Configuration conf HBaseConfiguration.create(getConf());String tablename args[0]; // 表名String targetTable args[1]; // 目标表// 获取到Job对象Job job Job.getInstance(conf);// 创建Scan对象Scan scan new Scan();// 通过Hbase工具类提交数据TableMapReduceUtil.initTableMapperJob(tablename, scan, MyMapper.class, Text.class, IntWritable.class, job);TableMapReduceUtil.initTableReducerJob(targetTable, MyReducer.class, job);// 开始提交数据job.waitForCompletion(true);return 0;/********** End *********/}}第2关HBase的MapReduce使用任务描述本关任务写一个HBase的MapReduce任务。相关知识为了完成本关任务你需要掌握编写自己的MapReduce。编写自己的MapReduceHBase结合MapReduce可以实现从HBase输入表到HBase输出表、从文件到HBase输出表、从HBase输入表到文件的几种场景。下面我们来完成一个从HBase输入表到HBase输出表的需求从HBase表里统计列族comment_info中列名content的字符串单词的出现次数并把统计结果存到HBase列族为word_info列名为count的另一个表里。我们分析一下需求统计的是字符串中单词的个数则key应该为Text类型value 为IntWritable类型输入表名为args[0]。Map节点配置如下String tablename args[0];String targetTable args[1];Scan scan new Scan();Job job Job.getInstance(conf);TableMapReduceUtil.initTableMapperJob(tablename,scan,MyMapper.class, Text.class, IntWritable.class,job);Map节点的目的是切分字符串获取单词名称把单词名称当做key值单词数1当成value值写到context.writekeyvalue方法中Map方法如下:public class MyMapper extends TableMapperText, IntWritable {private static byte[] family comment_info.getBytes();private static byte[] column content.getBytes();Overrideprotected void map(ImmutableBytesWritable rowKey, Result result, Context context)throws IOException, InterruptedException {byte[] value result.getValue(family, column);String content new String(value,utf-8);String[] split content.split(,);for(String str : split) {Text text new Text(str);IntWritable v new IntWritable(1);context.write(text,v);}}}}接下来我们配置及使用 Reduce 方法,输出表名为args[1], Reduce节点配置如下TableMapReduceUtil.initTableReducerJob(args[1],MyReducer.class,job);由于Reduce节点需要把统计结果放到Mutation中而Mutation是个抽象类一般都会使用继承它的Put类进行数据写入Reduce节点的使用方法如下public class MyReducer extends TableReducerText, IntWritable, ImmutableBytesWritable {private static byte[] family word_info.getBytes();private static byte[] column count.getBytes();Overridepublic void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException {int sum 0;for (IntWritable value : values) {sum value.get();}Put put new Put(Bytes.toBytes(key.toString()));put.addColumn(family,column,Bytes.toBytes(sum));context.write(null,put);}最后提交任务并阻塞等待job执行完成job.waitForCompletion();至此你就做完了你的第一个HBase的MapReduce任务啦编程要求根据提示在右侧Begin - End区域内补充代码完成一个从文件到HBase输出表的需求使用MapReduce统计单词个数。String[] args数组第一个参数为输入文件 第二个参数为输出表输入文件内容如下There will be no regret and sorrow if you fight with all your strengthgive a stranger one of your smiles. It might be the only sunshine he sees all dayVictory belongs to the most perseveringThe world is his who enjoys itwant it more that anythingLove is the greatest refreshment in lifeThe first step is as good as half overMy heart is with youCease to struggle and you cease to liveHe knows most who speaks leastUntil you make peace with who you are, you’ll never be content with what you haveOne needs 3 things to be truly happy living in the world: some thing to do, some one to love, some thing to hope forI lied when I said I didn’t like you. I lied when I said I didn’t care. I lie every time I try to tell myself I will never fall for you统计文件内容里出现的单词个数并把统计结果保存到列族为word_info单词个数为count的字段单词名称为rowkey的输出表里。测试说明补充完代码后点击测评平台会对你编写的代码进行测试当你的结果与预期输出一致时即为通过。测试输入t_comment t_word_count预期输出word:Iword_info:count 9word:beword_info:count 4word:isword_info:count 4word:theword_info:count 4word:toword_info:count 8word:withword_info:count 4word:youword_info:count 7开始你的任务吧祝你成功package com.processdata;import java.io.IOException;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.conf.Configured;import org.apache.hadoop.fs.Path;import org.apache.hadoop.hbase.HBaseConfiguration;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.TableReducer;import org.apache.hadoop.hbase.util.Bytes;import org.apache.hadoop.io.IntWritable;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.input.TextInputFormat;import org.apache.hadoop.util.Tool;/*** 词频统计**/public class WorldCountMapReduce2 extends Configured implements Tool {private static class MyMapper extends MapperObject, Text, Text, IntWritable {Overridepublic void map(Object object, Text value, Context context) throws IOException, InterruptedException {/********** Begin *********/// 根据题意,我们需要根据空格对指定数据进行拆分String[] split value.toString().split( );// 循环数组对值进行分类for (String str : split) {Text text new Text(str.getBytes());IntWritable v new IntWritable(1);context.write(text, v);}/********** End *********/}}private static class MyReducer extends TableReducerText, IntWritable, ImmutableBytesWritable {private static byte[] family word_info.getBytes();private static byte[] column count.getBytes();Overridepublic void reduce(Text key, IterableIntWritable values, Context context)throws IOException, InterruptedException {/********** Begin *********/int sum 0; // 用于统计// 循环Map中分类的值求和for (IntWritable value : values) {sum value.get();}// 将key和value进行聚和Put put new Put(Bytes.toBytes(key.toString()));put.addColumn(family, column, Bytes.toBytes(sum));// 通过文件方式将其输出context.write(null, put);/********** End *********/}}Overridepublic int run(String[] args) throws Exception {// 配置Job/********** Begin *********/// 配置Configuration conf HBaseConfiguration.create(getConf());String file args[0]; // 输入文件String targetTable args[1]; // 输出表Job job Job.getInstance(conf);// Map的Key的输入类型job.setMapOutputKeyClass(Text.class);// Map的Value的输入类型job.setMapOutputValueClass(IntWritable.class);// 需要执行的MapReduce类job.setJarByClass(WorldCountMapReduce2.class);// 文件输入格式FileInputFormat.addInputPath(job, new Path(file));// 设置Mapper类job.setMapperClass(MyMapper.class);// 开始执行任务TableMapReduceUtil.initTableReducerJob(targetTable, MyReducer.class, job);job.waitForCompletion(true);return 0;/********** End *********/}}有任何问题都可以随时关注私信