MapReduce实战练习: 使用搜狗实验数据对搜词热度排序（二）

我们接着上一篇《MapReduce实战练习: 使用搜狗实验数据对数据去重》 接着讲解关于对keyword（热搜词）的top100计算

###第一步，我们得到每个词的搜索次数

我们将上篇项目中的输出文件，作为本篇的输入源

#####计算每个关键词出现的次数，新建 KeyWordCount.java

源码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

package com.au.example;

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.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.FileInputFormat;  
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;  
import org.apache.hadoop.util.GenericOptionsParser;  

public class KeyWordCount {

	
	public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {  
        private final static IntWritable one = new IntWritable(1);  
          
        @Override  
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {  
            // 将输入的纯文本文件的数据转化成String  
            String line = value.toString();  
            // 将输入的数据首先按行进行分割  
            StringTokenizer tokenizerArticle = new StringTokenizer(line, "\n");  
            // 分别对每一行进行处理  
            while (tokenizerArticle.hasMoreElements()) {  
                // 每行按空格划分  
                StringTokenizer tokenizerLine = new StringTokenizer(tokenizerArticle.nextToken());  
                String c1 = tokenizerLine.nextToken();//  
                String c2 = tokenizerLine.nextToken();// 关键词  
                String c3 = tokenizerLine.nextToken();// 关键词  
                Text newline = new Text(c3);  
                context.write(newline, one);  
            }  
        }  
    }
	
	
	public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {  
        private IntWritable result =new IntWritable();  
          
        // 实现reduce函数  
        @Override  
        public void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {  
            int count = 0;  
             for(IntWritable val:values){  
                  count += val.get();  
             }  
            result.set(count);  
            context.write(key, result);  
        }  
    }  
	
	
	
	
	
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();  
        //设置输入输出文件目录  
        String[] ioArgs = new String[] { "hdfs://localhost:9000/clean_same_out", "hdfs://localhost:9000/Key_out" };  
        String[] otherArgs = new GenericOptionsParser(conf, ioArgs).getRemainingArgs();  
        if (otherArgs.length != 2) {  
            System.err.println("Usage:  <in> <out>");  
            System.exit(2);  
        }  
        //设置一个job  
        Job job = Job.getInstance(conf, "key Word count");  
        job.setJarByClass(KeyWordCount.class);  
          
        // 设置Map、Combine和Reduce处理类  
        job.setMapperClass(KeyWordCount.Map.class);  
        job.setCombinerClass(KeyWordCount.Reduce.class);  
        job.setReducerClass(KeyWordCount.Reduce.class);  
          
        // 设置输出类型  
        job.setOutputKeyClass(Text.class);  
        job.setOutputValueClass(IntWritable.class);  
          
        // 将输入的数据集分割成小数据块splites，提供一个RecordReder的实现  
        job.setInputFormatClass(TextInputFormat.class);  
          
        // 提供一个RecordWriter的实现，负责数据输出  
        job.setOutputFormatClass(TextOutputFormat.class);  
          
        // 设置输入和输出目录  
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));  
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));  
        System.exit(job.waitForCompletion(true) ? 0 : 1);  
		
	}

}


输出的数据文件，如下图：

###第二步，根据统计每一词的count，按照做从小到大的排序， 前100的热搜词

#####计算每个关键词出现的次数，新建 KeyWordCount.java

源码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
package com.autohome.example;

import java.io.IOException;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class TopKeyword {

	public static final int K = 100;

	public static class KMap extends Mapper<LongWritable, Text, IntWritable, Text> {  
        
        TreeMap<Integer, String> map = new TreeMap<Integer, String>();   
          
        @Override  
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {  
            String line = value.toString();  
            if(line.trim().length() > 0 && line.indexOf("\t") != -1) {  
                String[] arr = line.split("\t", 2);  
                String name = arr[0];  
                Integer num = Integer.parseInt(arr[1]);  
                map.put(num, name);  
                if(map.size() > K) {  
                    map.remove(map.firstKey());  
                }  
            }  
        }  
        
        @Override  
        protected void cleanup(Mapper<LongWritable, Text, IntWritable, Text>.Context context) throws IOException, InterruptedException {  
            for(Integer num : map.keySet()) {  
                context.write(new IntWritable(num), new Text(map.get(num)));  
            }  
        }  
        
    }  
        
	
	public static class KReduce extends Reducer<IntWritable, Text, IntWritable, Text> {  
        TreeMap<Integer, String> map = new TreeMap<Integer, String>();  
          
        @Override  
        public void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {  
            map.put(key.get(), values.iterator().next().toString());  
            if(map.size() > K) {  
                map.remove(map.firstKey());  
            }  
        }  
  
        @Override  
        protected void cleanup(Reducer<IntWritable, Text, IntWritable, Text>.Context context) throws IOException, InterruptedException {  
            for(Integer num : map.keySet()) {  
                context.write(new IntWritable(num), new Text(map.get(num)));  
            }  
        }  
    }  
	
	
	
	
	public static void main(String[] args) throws Exception {
		
		Configuration conf = new Configuration();  
        //设置输入输出文件目录  
        String[] ioArgs = new String[] { "hdfs://localhost:9000/Key_out", "hdfs://localhost:9000/top_out" };  
        String[] otherArgs = new GenericOptionsParser(conf, ioArgs).getRemainingArgs();  
        if (otherArgs.length != 2) {  
            System.err.println("Usage:  <in> <out>");  
            System.exit(2);  
        }  
        //设置一个job  
        Job job = Job.getInstance(conf, "top Keyword");  
          
        job.setJarByClass(TopKeyword.class);  
          
        // 设置Map、Combine和Reduce处理类  
        job.setMapperClass(KMap.class);  
        job.setCombinerClass(KReduce.class);  
        job.setReducerClass(KReduce.class);  
          
        // 设置输出类型  
        job.setOutputKeyClass(IntWritable.class);  
        job.setOutputValueClass(Text.class);  
          
        // 将输入的数据集分割成小数据块splites，提供一个RecordReder的实现  
        job.setInputFormatClass(TextInputFormat.class);  
          
        // 提供一个RecordWriter的实现，负责数据输出  
        job.setOutputFormatClass(TextOutputFormat.class);  
          
        // 设置输入和输出目录  
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));  
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));  
        System.exit(job.waitForCompletion(true) ? 0 : 1);  

	}

}

输出结果为：

我们可以看到前面是搜索的次数， 后面是搜索的关键词， 按照搜索的热度，取最高的100名， 按从小到大的顺序。

赏

谢谢你请我吃糖果

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

很惭愧<br><br>只做了一点微小的工作<br>谢谢大家