InAction-MR的topK

本来只是想拿搜狗的数据练练手的，却无意踏足MR的topK问题。经过几番波折，虽然现在看起来很简单，但是摸爬滚打中也学到了不少

数据是搜狗实验室下的搜索日志，格式大概为：

 00:00:00    2982199073774412    [360安全卫士]    8 3    download.it.com.cn/softweb/software/firewall/antivirus/20067/17938.html

 00:00:00    07594220010824798    [哄抢救灾物资]    1 1    news.21cn.com/social/daqian/2008/05/29/4777194_1.shtml

 00:00:00    5228056822071097    [75810部队]    14 5    www.greatoo.com/greatoo_cn/list.asp?link_id=276&title=%BE%DE%C2%D6%D0%C2%CE%C5

 00:00:00    6140463203615646    [绳艺]    62 36    www.jd-cd.com/jd_opus/xx/200607/706.html

 00:00:00    8561366108033201    [汶川地震原因]    3 2    www.big38.net/

 00:00:00    23908140386148713    [莫衷一是的意思]    1 2    www.chinabaike.com/article/81/82/110/2007/2007020724490.html

 00:00:00    1797943298449139    [星梦缘全集在线观看]    8 5    www.6wei.net/dianshiju/????\xa1\xe9|????do=index

 00:00:00    00717725924582846    [闪字吧]    1 2    www.shanziba.com/

我只是要搜索词，其他的不管，然后通过MR计算出搜索量最高的前N个词（N自定义）

整体项目结构为：

InAction-MR的topK
先来个类处理根据日志格式拿出搜索词

SEA.java

 package org.admln.topK;

 /**

  * @author admln

  *

  */

 public class SEA {

     private String seaWord;

     private boolean isValid;

     public static SEA parser(String line) {

         SEA sea = new SEA();

         String str = line.split("\t")[2];

         if(str.length()<3) {

             sea.setValid(false);

         }else {

             sea.setValid(true);

             sea.setSeaWord(str.substring(1, str.length()-1));

         }

         return sea;

     }

     public String getSeaWord() {

         return seaWord;

     }

     public void setSeaWord(String seaWord) {

         this.seaWord = seaWord;

     }

     public boolean isValid() {

         return isValid;

     }

     public void setValid(boolean isValid) {

         this.isValid = isValid;

     }

 }

然后就是MR

 package org.admln.topK;

 import java.io.IOException;

 import java.util.Collections;

 import java.util.Map.Entry;

 import java.util.Set;

 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.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.output.FileOutputFormat;

 /**

  * @author admln

  *

  */

 public class TopK {

     public static class topKMapper extends

             Mapper<Object, Text, Text, IntWritable> {

         Text word = new Text();

         IntWritable ONE = new IntWritable(1);

         @Override

         public void map(Object key, Text value, Context context)

                 throws IOException, InterruptedException {

             SEA sea = SEA.parser(value.toString());

             if (sea.isValid()) {

                 word.set(sea.getSeaWord());

                 context.write(word, ONE);

             }

         }

     }

     public static class topKReducer extends

             Reducer<Text, IntWritable, Text, IntWritable> {

         int sum;

         int max;

         private static TreeMap<Integer,String> tree = new TreeMap<Integer,String>(Collections.reverseOrder());

         public void reduce(Text key, Iterable<IntWritable> values,

                 Context context) {

             sum = 0;

             max = context.getConfiguration().getInt("topK", 10);

             for (IntWritable val : values) {

                 sum += val.get();

             }

             tree.put(Integer.valueOf(sum), key.toString());

             if (tree.size() > max) {

                 tree.remove(tree.lastKey());

             }

         }

         @Override

         protected void cleanup(Context context) throws IOException, InterruptedException {

             Set<Entry<Integer, String>> set = tree.entrySet();

             for (Entry<Integer, String> entry : set) {

                 context.write(new Text(entry.getValue()), new IntWritable(entry.getKey()));

             }

         }

     }

     public static void main(String[] args) throws Exception {

         Path input = new Path("hdfs://hadoop:8020/input/topK/");

         Path output = new Path("hdfs://hadoop:8020/output/topK/");

         Configuration conf = new Configuration();

         conf.setInt("topK", Integer.valueOf(args[1]));

         Job job = new Job(conf, "topK");

         job.setJarByClass(TopK.class);

         job.setMapperClass(topKMapper.class);

         job.setReducerClass(topKReducer.class);

         job.setOutputKeyClass(Text.class);

         job.setOutputValueClass(IntWritable.class);

         FileInputFormat.addInputPath(job, input);

         FileOutputFormat.setOutputPath(job, output);

         System.exit(job.waitForCompletion(true) ? 0 : 1);

     }

 }