项目整体介绍

对类似WordCount案例的词频统计，并将统计结果按出现次数降序排列。

网上有很多帖子，均用的相似方案，重写某某方法然后。。。运行起来可能会报这样那样的错误，这里实现了一种解决方案，分享出来供大家参考：编写两个MapReduce程序，第一个程序进行词频统计，第二个程序进行降序处理，由于是降序，还需要自定义对象，在对象内部实现降序排序。

一、项目背景及数据集说明

现有某电商网站用户对商品的收藏数据，记录了用户收藏的商品id以及收藏日期，名为buyer_favorite1。buyer_favorite1包含：买家id，商品id，收藏日期这三个字段，数据以“\t”分割，样例展现如下：

二、编写MapReduce程序，统计每个买家收藏商品数量。（即统计买家id出现的次数）

前置说明

1.配置好Hadoop集群环境，并开启相应服务、
2.在hdfs对应路径上先上传好文件，可以自己根据文件路径定义，这里是"hdfs://localhost:9000/mymapreduce1/in/buyer_favorite1"。同时再定义好输出路径
3.这里是整个程序（词频降序）的入口，若只是想统计词频，请注释掉WordCountSortDESC.mainJob2();

package mapreduce;import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class WordCount {public static void main(String[] args) {Configuration conf = new Configuration();conf.set("yarn,resourcemanager", "bym@d2e674ec1e78");try {Job job = Job.getInstance(conf, "111");job.setJobName("WordCount");job.setJarByClass(WordCount.class);job.setMapperClass(doMapper.class); // 这里就是设置下job使用继承doMapper类，与定义的内容保持一致job.setReducerClass(doReducer.class); // 同上，设置Reduce类型job.setMapOutputKeyClass(Text.class); // 如果map的输出和reduce的输出不一样，这里要分别定义好格式job.setMapOutputValueClass(IntWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);Path in = new Path("hdfs://localhost:9000/mymapreduce1/in/buyer_favorite1");Path out = new Path("hdfs://localhost:9000/mymapreduce1/out");FileInputFormat.addInputPath(job, in);FileOutputFormat.setOutputPath(job, out);if (job.waitForCompletion(true)) {System.out.println("WordCount completition");WordCountSortDESC.mainJob2();System.out.println("diaoyong");}} catch (Exception e) {e.printStackTrace();}// System.exit(job.waitForCompletion(true) ? 0 : 1);}// 第一个Object表示输入key的类型、是该行的首字母相对于文本文件的首地址的偏移量;// 第二个Text表示输入value的类型、存储的是文本文件中的一行（以回车符为行结束标记）;// 第三个Text表示输出键的类型；第四个IntWritable表示输出值的类型public static class doMapper extendsMapper<LongWritable, Text, Text, IntWritable> {public static final IntWritable one = new IntWritable(1);public static Text word = new Text();@Override// 前面两个Object key,Text value就是输入的key和value，第三个参数Context// context是可以记录输入的key和value。protected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {// StringTokenizer是Java工具包中的一个类，用于将字符串进行拆分StringTokenizer tokenizer = new StringTokenizer(value.toString(),"\t");// 返回当前位置到下一个分隔符之间的字符串, 并把字符串设置成Text格式word.set(tokenizer.nextToken());context.write(word, one);}}// 参数依次表示是输入键类型，输入值类型，输出键类型，输出值类型public static class doReducer extendsReducer<Text, IntWritable, Text, Text> {@Override// 输入的是键值类型，其中值类型为归并后的结果，输出结果为Context类型protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {int sum = 0;for (IntWritable value : values) {sum += value.get();}context.write(key, new Text(Integer.toString(sum)));}}
}

三、核心问题：再次编写MapReduce程序，将上一步统计的结果降序排列

前置说明

1.这里将上一步统计的结果作为输入，进行第二次mapreduce程序的运行。因此要注意输入路径与上一步的输出路径保持一致。
2.由于是降序排列，只能自定义FlowBean对象，内部实现排序方式。否则，升序可以利用shuffle机制默认的排序策略不用自定义对象排序，这里不再叙述。

package mapreduce;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class WordCountSortDESC {public static void mainJob2() {Configuration conf = new Configuration();conf.set("yarn,resourcemanager", "bym@d2e674ec1e78");try {Job job = Job.getInstance(conf, "1111");job.setJobName("WordCountSortDESC");job.setJarByClass(WordCountSortDESC.class);job.setMapperClass(TwoMapper.class); // 这里就是设置下job使用继承doMapper类，与定义的内容保持一致job.setReducerClass(TwoReducer.class); // 同上，设置Reduce类型job.setMapOutputKeyClass(FlowBean.class);job.setMapOutputValueClass(Text.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);Path in = new Path("hdfs://localhost:9000/mymapreduce1/out");Path out = new Path("hdfs://localhost:9000/mymapreduce1/out555");FileInputFormat.addInputPath(job, in);FileOutputFormat.setOutputPath(job, out);if (job.waitForCompletion(true)) {System.out.println("DESC Really Done");}} catch (Exception e) {System.out.println("errormainJob2-----------");}}public static class TwoMapper extends Mapper<Object, Text, FlowBean, Text> {private FlowBean outK = new FlowBean();private Text outV = new Text();@Overrideprotected void map(Object key, Text value, Context context)throws IOException, InterruptedException {// 由于真实的数据存储在文件块上，这里是因为数据量较小，可以保证只在一个文件块FileSplit fs = (FileSplit) context.getInputSplit();if (fs.getPath().getName().contains("part-r-00000")) {// 1 获取一行数据String line = value.toString();// 2 按照"\t",切割数据String[] split = line.split("\t");// 3 封装outK outVoutK.setNumber(Long.parseLong(split[1]));outV.set(split[0]);// 4 写出outK outVcontext.write(outK, outV);} else {System.out.println("error-part-r-------------------");}}}public static class TwoReducer extendsReducer<FlowBean, Text, Text, FlowBean> {@Overrideprotected void reduce(FlowBean key, Iterable<Text> values,Context context) throws IOException, InterruptedException {// 遍历values集合,循环写出,避免总流量相同的情况for (Text value : values) {// 调换KV位置,反向写出context.write(value, key);}}}public static class FlowBean implements WritableComparable<FlowBean> {private long number;// 提供无参构造public FlowBean() {}public long getNumber() {return number;}public void setNumber(long number) {this.number = number;}// 实现序列化和反序列化方法,注意顺序一定要一致@Overridepublic void write(DataOutput out) throws IOException {out.writeLong(this.number);}@Overridepublic void readFields(DataInput in) throws IOException {this.number = in.readLong();}@Overridepublic String toString() {return number + "\t";}@Overridepublic int compareTo(FlowBean o) {// 按照总流量比较,倒序排列if (this.number > o.number) {return -1;} else if (this.number < o.number) {return 1;} else {return 0;}}}}

四、结果展示：

执行查看文件命令

hadoop fs -cat /mymapreduce1/out555/part-r-00000

可以发现已经进行了降序排列，其他数据集结果应类似。