Hadoop：MapReduce编程之统计单词的数目

要求&＃xff1a;统计出每个单词的数目&＃xff0c;显示结果为单词 单词的长度 单词的数目

分析&＃xff1a;由于MapReduce中的数据传输只能以形式传输&＃xff0c;只能传递两列数据&＃xff0c;因此为了实现三列数据的传输&＃xff0c;我们可以通过两种方法解决&＃xff1a;字符串拼接和自定义数据类型封装JavaBean。

一、字符串拼接

我们可以将单词与单词对应的长度拼接成一列数据进行传输
代码实现&＃xff1a;

package com.miao.wordcount;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.io.IntWritable;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;import java.io.IOException;/*** &＃64;ClassName WordConcat* &＃64;Description TODO 统计每个单词出现的次数&＃xff0c;并且单词与对应长度拼接* &＃64;Date 2021-04-27 20:08:50* &＃64;Create By Miao*/
public class WordConcat extends Configured implements Tool {public int run(String[] args) throws Exception {//构建JobJob job &＃61; Job.getInstance(this.getConf(),"wordConcat");job.setJarByClass(WordConcat.class);//配置Jobjob.setInputFormatClass(TextInputFormat.class);//指定输入源TextInputFormat.setInputPaths(job,new Path("D:\\Study\\idea\\MavenProject\\count.txt"));job.setMapperClass(WCMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setReducerClass(WCReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);job.setOutputFormatClass(TextOutputFormat.class);//指定输出源Path outputPath &＃61; new Path("D:\\Study\\idea\\MavenProject\\output\\five");FileSystem fs &＃61; FileSystem.get(this.getConf());if(fs.exists(outputPath)){fs.delete(outputPath,true);}TextOutputFormat.setOutputPath(job,outputPath);//提交Jobreturn job.waitForCompletion(true) ? 0 : -1;}public static void main(String[] args) throws Exception {Configuration conf &＃61; new Configuration();int status &＃61; ToolRunner.run(conf, new WordConcat(), args);System.exit(status);}public static class WCMapper extends Mapper {//输出的Key2Text outputKey &＃61; new Text();//输出的Value2IntWritable outputValue &＃61; new IntWritable(1);/*** 每条KV调用一次map* &＃64;param key&＃xff1a;行的偏移量* &＃64;param value&＃xff1a;行的内容* &＃64;param context* &＃64;throws IOException* &＃64;throws InterruptedException*/&＃64;Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//将每行的内容分割得到每个单词String[] words &＃61; value.toString().split("\\s&＃43;");//迭代取出每个单词作为Key2for (String word : words) {//将当前的单词作为Key2this.outputKey.set(word);//将Key2和Value2传递到下一步context.write(outputKey,outputValue);}}}public static class WCReducer extends Reducer {//输出Key3:拼接单词和单词的长度Text outputKey &＃61; new Text();//输出Value3IntWritable outputValue &＃61; new IntWritable();/*** 每一组调用一次* &＃64;param key&＃xff1a;单词* &＃64;param values&＃xff1a;所有相同单词对应的1* &＃64;param context* &＃64;throws IOException* &＃64;throws InterruptedException*/&＃64;Overrideprotected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {int sum &＃61; 0;for (IntWritable value : values) {//取出当前单词所有 1&＃xff0c;进行累加sum &＃43;&＃61; value.get();}//给key3赋值,单词和单词的长度this.outputKey.set(key.toString()&＃43;"\t"&＃43;key.toString().length());//给Value3赋值this.outputValue.set(sum);//传递到下一步context.write(outputKey,this.outputValue);}}
}


二、自定义数据类型封装JavaBean

我们可以通过定义一个类型&＃xff0c;存储一个String属性和一个Int属性&＃xff0c;用于实现MapReduce中的三列数据的传输

代码实现&＃xff1a;

自定义数据类型&＃xff1a;

package com.miao.bean;import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;public class WordCountBean implements Writable {private String firstKey;private int secondkey;public WordCountBean() {}public String getFirstKey() {return firstKey;}public void setFirstKey(String firstKey) {this.firstKey &＃61; firstKey;}public int getSecondkey() {return secondkey;}public void setSecondkey(int secondkey) {this.secondkey &＃61; secondkey;}&＃64;Overridepublic String toString() {return this.firstKey &＃43; "\t" &＃43; this.secondkey;}&＃64;Overridepublic void write(DataOutput dataOutput) throws IOException {dataOutput.writeUTF(this.firstKey);dataOutput.writeInt(this.secondkey);}&＃64;Overridepublic void readFields(DataInput dataInput) throws IOException {this.firstKey &＃61; dataInput.readUTF();this.secondkey &＃61; dataInput.readInt();}
}

测试&＃xff1a;

package com.miao.wordcount;import com.miao.bean.WordCountBean;
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.io.IntWritable;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;import java.io.IOException;public class WordCount extends Configured implements Tool {public int run(String[] args) throws Exception {//构建JobJob job &＃61; Job.getInstance(this.getConf(),"wordConcat");job.setJarByClass(WordCount.class);//配置Jobjob.setInputFormatClass(TextInputFormat.class);TextInputFormat.setInputPaths(job,new Path("D:\\Study\\idea\\MavenProject\\count.txt"));job.setMapperClass(WCMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setReducerClass(WCReducer.class);job.setOutputKeyClass(WordCountBean.class);job.setOutputValueClass(IntWritable.class);job.setOutputFormatClass(TextOutputFormat.class);Path outputPath &＃61; new Path("D:\\Study\\idea\\MavenProject\\output\\five");FileSystem fs &＃61; FileSystem.get(this.getConf());if(fs.exists(outputPath)){fs.delete(outputPath,true);}TextOutputFormat.setOutputPath(job,outputPath);//提交Jobreturn job.waitForCompletion(true) ? 0 : -1;}public static void main(String[] args) throws Exception {Configuration conf &＃61; new Configuration();int status &＃61; ToolRunner.run(conf, new WordCount(), args);System.exit(status);}public static class WCMapper extends Mapper {//输出的Key2Text outputKey &＃61; new Text();//输出的Value2IntWritable outputValue &＃61; new IntWritable(1);/*** 每条KV调用一次map* &＃64;param key&＃xff1a;行的偏移量* &＃64;param value&＃xff1a;行的内容* &＃64;param context* &＃64;throws IOException* &＃64;throws InterruptedException*/&＃64;Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//将每行的内容分割得到每个单词String[] words &＃61; value.toString().split("\\s&＃43;");//迭代取出每个单词作为Key2for (String word : words) {//将当前的单词作为Key2this.outputKey.set(word);//将Key2和Value2传递到下一步context.write(outputKey,outputValue);}}}public static class WCReducer extends Reducer {//输出Key3:拼接单词和单词的长度WordCountBean outputKey&＃61;new WordCountBean();//输出Value3IntWritable outputValue &＃61; new IntWritable();/*** 每一组调用一次* &＃64;param key&＃xff1a;单词* &＃64;param values&＃xff1a;所有相同单词对应的1* &＃64;param context* &＃64;throws IOException* &＃64;throws InterruptedException*/&＃64;Overrideprotected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {int sum &＃61; 0;for (IntWritable value : values) {//取出当前单词所有 1&＃xff0c;进行累加sum &＃43;&＃61; value.get();}//给key3赋值,单词和单词的长度this.outputKey.setFirstKey(key.toString());this.outputKey.setSecondkey(key.toString().length());//给Value3赋值this.outputValue.set(sum);//传递到下一步context.write(this.outputKey,this.outputValue);}}
}


运行结果&＃xff1a;

三、自定义数据类型实现比较器接口

当WordCountBean这个类型数据作为key2时需要实现比较器接口

代码实现&＃xff1a;

package com.miao.bean;import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;public class WCBean2 implements WritableComparable {private String firstKey;private int secondKey;public WCBean2(){}public String getFirstKey() {return firstKey;}public void setFirstKey(String firstKey) {this.firstKey &＃61; firstKey;}public int getSecondKey() {return secondKey;}public void setSecondKey(int secondKey) {this.secondKey &＃61; secondKey;}&＃64;Overridepublic String toString() {return this.firstKey&＃43;"\t"&＃43;this.secondKey;}public void write(DataOutput out) throws IOException {out.writeUTF(this.firstKey);//类型必须一致out.writeInt(this.secondKey);}//反序列化public void readFields(DataInput in) throws IOException {this.firstKey &＃61; in.readUTF(); //必须与序列化的顺序保持一致this.secondKey &＃61; in.readInt();}/*** 当这个类型作为K2经过shuffle时&＃xff0c;需要调用该方法进行比较判断&＃xff1a;排序和分组* &＃64;param o* &＃64;return*/public int compareTo(WCBean2 o) {//先比较第一个属性int comp &＃61; this.getFirstKey().compareTo(o.getFirstKey());//如果第一个值&＃xff0c;比较第二个值&＃xff0c;用第二个属性的比较的结果作为最后的结果if(comp &＃61;&＃61; 0){return Integer.valueOf(this.getSecondKey()).compareTo(Integer.valueOf(o.getSecondKey()));}//默认升序&＃xff0c;如果要降序&＃xff0c;添加负号即可return -comp;}
}

测试&＃xff1a;

package com.miao.bean;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.io.IntWritable;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import java.io.IOException;public class WordCount2 extends Configured implements Tool {public int run(String[] args) throws Exception {//构建JobJob job &＃61; Job.getInstance(this.getConf(),"wordConcat");job.setJarByClass(WordCount2.class);//配置Jobjob.setInputFormatClass(TextInputFormat.class);//使用程序的第一个参数作为输入TextInputFormat.setInputPaths(job,new Path("D:\\Study\\idea\\MavenProject\\count.txt"));job.setMapperClass(WCMapper.class);job.setMapOutputKeyClass(WCBean2.class);job.setMapOutputValueClass(IntWritable.class);job.setReducerClass(WCReducer.class);job.setOutputKeyClass(WCBean2.class);job.setOutputValueClass(IntWritable.class);job.setOutputFormatClass(TextOutputFormat.class);//使用程序的第二个参数作为输出路径Path outputPath &＃61; new Path("D:\\Study\\idea\\MavenProject\\output\\seven");FileSystem fs &＃61; FileSystem.get(this.getConf());if(fs.exists(outputPath)){fs.delete(outputPath,true);}TextOutputFormat.setOutputPath(job,outputPath);//提交Jobreturn job.waitForCompletion(true) ? 0 : -1;}public static void main(String[] args) throws Exception {Configuration conf &＃61; new Configuration();int status &＃61; ToolRunner.run(conf, new WordCount2(), args);System.exit(status);}public static class WCMapper extends Mapper {//输出的Key2WCBean2 outputKey&＃61;new WCBean2();//输出的Value2IntWritable outputValue &＃61; new IntWritable(1);&＃64;Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//将每行的内容分割得到每个单词String[] words &＃61; value.toString().split("\\s&＃43;");//迭代取出每个单词作为Key2for (String word : words) {//将当前的单词作为Key2this.outputKey.setFirstKey(word.toString());this.outputKey.setSecondKey(word.toString().length());//将Key2和Value2传递到下一步context.write(outputKey,outputValue);}}}public static class WCReducer extends Reducer {//输出Value3IntWritable outputValue &＃61; new IntWritable();&＃64;Overrideprotected void reduce(WCBean2 key, Iterable values, Context context) throws IOException, InterruptedException {int sum &＃61; 0;for (IntWritable value : values) {//取出当前单词所有 1&＃xff0c;进行累加sum &＃43;&＃61; value.get();}//给Value3赋值this.outputValue.set(sum);//传递到下一步context.write(key,this.outputValue);}}
}


四、自定义排序器

默认排序方式为升序&＃xff0c;如果想要实现降序排序&＃xff0c;我们需要自定义排序器

package com.miao.sort;import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;public class WCSort extends WritableComparator {//step1&＃xff1a;注册public WCSort(){super(Text.class,true);}//step2&＃xff1a;实现比较&＃64;Overridepublic int compare(WritableComparable a, WritableComparable b) {//将两个比较器对象强转为要比较的Text类型Text t1 &＃61; (Text) a;Text t2 &＃61; (Text) b;//实现两个Text类型的比较&＃xff0c;降序排序return -t1.toString().compareTo(t2.toString());}
}


最后需要在测试类中添加&＃xff1a;

job.setSortComparatorClass(WCSort.class); //指定排序器