真正帮你实现—MapReduce统计WordCount词频，并将统计结果按出现次数降序排列

骑猪看日落 2023-10-14 10:02 61阅读 0赞

## 项目整体介绍 ##

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

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

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

现有某电商网站用户对商品的收藏数据，记录了用户收藏的商品id以及收藏日期，名为buyer\_favorite1。buyer\_favorite1包含：买家id，商品id，收藏日期这三个字段，数据以“\\t”分割，样例展现如下：![在这里插入图片描述][eb4d09e557914711bca2c8018b7d6156.png]

## 二、编写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 extends
    			Mapper<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 extends
    			Reducer<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();
    
    		@Override
    		protected 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 outV
    				outK.setNumber(Long.parseLong(split[1]));
    				outV.set(split[0]);
    
    				// 4 写出outK outV
    				context.write(outK, outV);
    			} else {
        
    				System.out.println("error-part-r-------------------");
    			}
    		}
    	}
    
    	public static class TwoReducer extends
    			Reducer<FlowBean, Text, Text, FlowBean> {
        
    		@Override
    		protected 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;
    		}
    
    		// 实现序列化和反序列化方法,注意顺序一定要一致
    		@Override
    		public void write(DataOutput out) throws IOException {
        
    			out.writeLong(this.number);
    		}
    
    		@Override
    		public void readFields(DataInput in) throws IOException {
        
    			this.number = in.readLong();
    		}
    
    		@Override
    		public String toString() {
        
    			return number + "\t";
    		}
    
    		@Override
    		public 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

![在这里插入图片描述][b6740dc957714d259ec49a6ee1276522.png]  
可以发现已经进行了降序排列，其他数据集结果应类似。

[eb4d09e557914711bca2c8018b7d6156.png]: https://img-blog.csdnimg.cn/eb4d09e557914711bca2c8018b7d6156.png
[b6740dc957714d259ec49a6ee1276522.png]: https://img-blog.csdnimg.cn/b6740dc957714d259ec49a6ee1276522.png