[MapReduce编程]用MapReduce大刀砍掉海量数据离线处理问题

这篇文章我以前是拜读过的，今天闲来没事，就想拿来当作MapReduce的练习。

MapReduce这把刀太大，刀大了问题就抵不住这刀锋了，事实上一开始我想着，这么多些题目，当是要花很多功夫的，但当我作完一题继续看下面的题目的时候，才发现这些题目在MapReduce模型下显得大同小异了，看来拿大刀的人是无论砍的是木头仍是人头的，而是直接抽象成柱形物而后抡起刀一刀就下去了。

  直入主题：

  一、海量日志数据，提取出某日访问百度次数最多的前K个IP。[稍微改变]

                说明：每一次访问网页就在日志中记录1次访问者的IP，独占一行，一个小数据能够在这里下载。

    实在是想不出如何能在一个Job中解决这个问题，因此仍是把它拉扯成了两个Job来解决。

      Job1：将相同IP的记录合并，造成<ip,count>形式,其中count是对这个ip的计数。

      Job2：按count排序<ip,count>并选择前K个进行输出。

    这里我写了一个可序列化的类IPAndCount，若是稍微熟悉MapReduce或者看明白我以前写的 关系型MapReduce模式：选择、分组和组内排序 你就知道这是为了排序而准备的。MapReduce有一个“Shuffle and sort”，这个阶段是利用key来对tuple进行排序的，而排序时调用的即是key的compareTo()方法。事实上若是job1输出的数据两足够小，咱们彻底能够在内存中进行排序而利用MapReduce框架，这样就能够省下一个Reduce阶段，可是对于这个问题显然不行。

    IPAndCount很直白，就是包装了上述的<ip,count>。

[java]  view plain copy print ?

1. import java.io.DataInput;  
2. import java.io.DataOutput;  
3. import java.io.IOException;  
4.   
5. import org.apache.hadoop.io.IntWritable;  
6. import org.apache.hadoop.io.Text;  
7. import org.apache.hadoop.io.WritableComparable;  
8.   
9.   
10. public class IPAndCount implements WritableComparable{  
11.     Text ip;  
12.     IntWritable count;  
13.       
14.     public IPAndCount(){  
15.         this.ip = new Text("");  
16.         this.count = new IntWritable(1);  
17.     }  
18.       
19.     public IPAndCount(Text ip, IntWritable count){  
20.         this.ip = ip;  
21.         this.count = count;  
22.     }  
23.       
24.     public IPAndCount(String ip, int count){  
25.         this.ip = new Text(ip);  
26.         this.count = new IntWritable(count);  
27.     }  
28.       
29.     public void readFields(DataInput in) throws IOException {  
30.         ip.readFields(in);  
31.         count.readFields(in);  
32.     }  
33.   
34.     public void write(DataOutput out) throws IOException {  
35.         ip.write(out);  
36.         count.write(out);  
37.     }  
38.   
39.     public int compareTo(Object o) {  
40.         return ((IPAndCount)o).count.compareTo(count) == 0?   
41.                 ip.compareTo(((IPAndCount)o).ip):((IPAndCount)o).count.compareTo(count);//若是只比较count会丢失数据，应该是suffle阶段的问题  
42.     }  
43.       
44.     public int hashCode(){  
45.         return ip.hashCode();  
46.     }  
47.       
48.     public boolean equals(Object o){  
49.         if(!(o instanceof IPAndCount))  
50.             return false;  
51.         IPAndCount other = (IPAndCount)o;  
52.         return ip.equals(other.ip) && count.equals(other.count);  
53.     }  
54.       
55.     public String toString(){  
56.         StringBuffer buf = new StringBuffer("[ip=");  
57.         buf.append(ip.toString());  
58.         buf.append(",count=");  
59.         buf.append(count.toString());  
60.         buf.append("]");  
61.         return buf.toString();  
62.     }  
63.       
64.     public Text getIp() {  
65.         return ip;  
66.     }  
67.     public void setIp(Text ip) {  
68.         this.ip = ip;  
69.     }  
70.     public IntWritable getCount() {  
71.         return count;  
72.     }  
73.     public void setCount(IntWritable count) {  
74.         this.count = count;  
75.     }  
76. }  

下面对FindActiveIp进行说明：

SumUpIpMapper和SumUpIPReducer事实上就是一个MapReduce中最基础的词频统计程序WordCount。你能够加一个Combiner来优化一下，我遗漏了。

从配置中能够看见两个Job的配置：job 和job2。

依赖关系是job -> job2，代码中使用了JobControl来解决做业间的依赖关系，JobControl.run()方法会在做业都运行完后才返回。

Job2的输入路径是Job1的输出路径，从参数中能够看出这一点。

Job1的输出在输出文件中的表现是：ip,count
Job2再从文件中读入，使用的是KeyValueTextInputFormat，它对应的是TextOutputFormat，咱们能够从job1的配置中看出来。

BeforeSortIPMapper从job1的输出中读取数据并包装成IPAndCount类型，以便MapReduce框架在“shuffle and sort”阶段利用它来排序。

最后SelectTopKIPReducer选出前K个进行输出便可，在这里咱们设置最后的reduce只有一个reduce task，以使全部数据汇聚到一台机子上进行处理。

[java]  view plain copy print ?

1. import java.io.IOException;  
2.   
3. import org.apache.hadoop.conf.Configuration;  
4. import org.apache.hadoop.conf.Configured;  
5. import org.apache.hadoop.fs.Path;  
6. import org.apache.hadoop.io.IntWritable;  
7. import org.apache.hadoop.io.LongWritable;  
8. import org.apache.hadoop.io.Text;  
9. import org.apache.hadoop.mapred.lib.ChainMapper;  
10. import org.apache.hadoop.mapreduce.Job;  
11. import org.apache.hadoop.mapreduce.Mapper;  
12. import org.apache.hadoop.mapreduce.Reducer;  
13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
14. import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;  
15. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
16. import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;  
17. import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;  
18. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
19. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;  
20. import org.apache.hadoop.util.Tool;  
21. import org.apache.hadoop.util.ToolRunner;  
22.   
23.   
24. public class FindActiveIP extends Configured implements Tool{  
25.       
26.     public static class SumUpIPMapper extends Mapper<LongWritable,Text,Text,IntWritable>{  
27.         IntWritable  one = new IntWritable(1);  
28.         public void map(LongWritable key, Text value, Context context)   
29.             throws IOException,InterruptedException{  
30.             context.write(value, one);  
31.         }  
32.     }  
33.   
34.     public static class SumUpIPReducer extends Reducer<Text,IntWritable,Text,IntWritable>{  
35.         //这里能够选择前k个进行输出以优化  
36.         public void reduce(Text key, Iterable<IntWritable> values, Context context)  
37.             throws IOException, InterruptedException{  
38.             int sum = 0;  
39.             for(IntWritable v : values){  
40.                 sum += v.get();  
41.             }  
42.             context.write(key, new IntWritable(sum));  
43.         }  
44.     }  
45.       
46.       
47.     public static class BeforeSortIPMapper extends Mapper<Text,Text,IPAndCount,Text>{  
48.         public void map(Text key, Text value, Context context)  
49.             throws IOException,InterruptedException{  
50.             IPAndCount tmp = new IPAndCount(key,new IntWritable(Integer.valueOf(value.toString())));  
51.             System.out.println(tmp);  
52.             context.write(tmp,new Text());  
53.         }  
54.     }  
55.       
56.       
57.     //set num of this reducer to one  
58.     public static class SelectTopKIPReducer extends Reducer<IPAndCount,Text,IPAndCount,Text>{  
59.         int counter = 0;  
60.         int K = 10;  
61.         public void reduce(IPAndCount key, Iterable<Text> values, Context context)  
62.             throws IOException, InterruptedException{  
63.             System.out.println(key);  
64.             if(counter < K){  
65.                 context.write(key, null);  
66.                   
67.                 counter++;  
68.             }  
69.               
70.         }  
71.     }  
72.     public int run(String[] args) throws Exception {  
73.         Configuration conf = new Configuration();  
74.         Job job = new Job(conf,"SumUpIP");  
75.         job.setJarByClass(FindActiveIP.class);  
76.         job.setInputFormatClass(TextInputFormat.class);  
77.         job.setOutputFormatClass(TextOutputFormat.class);  
78.         job.getConfiguration().set("mapred.textoutputformat.separator", ",");  
79.         Path in = new Path(args[0]);  
80.         Path out = new Path(args[1]);  
81.         FileInputFormat.setInputPaths(job, in);  
82.         FileOutputFormat.setOutputPath(job, out);  
83.         job.setMapperClass(SumUpIPMapper.class);  
84.         job.setReducerClass(SumUpIPReducer.class);  
85.         job.setMapOutputKeyClass(Text.class);  
86.         job.setMapOutputValueClass(IntWritable.class);  
87.         job.setOutputKeyClass(Text.class);  
88.         job.setOutputValueClass(IntWritable.class);  
89.         job.setNumReduceTasks(7);  
90.           
91.         Configuration conf2 = new Configuration();  
92.         Job job2 = new Job(conf2,"SortAndFindTopK");  
93.         job2.setJarByClass(FindActiveIP.class);  
94.         job2.setInputFormatClass(KeyValueTextInputFormat.class);  
95.         job2.getConfiguration().set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ",");  
96.         job2.setOutputFormatClass(TextOutputFormat.class);  
97.         Path in2 = new Path(args[1]);  
98.         Path out2 = new Path(args[2]);  
99.         FileInputFormat.setInputPaths(job2,in2);  
100.         FileOutputFormat.setOutputPath(job2, out2);  
101.         job2.setMapperClass(BeforeSortIPMapper.class);  
102.         job2.setReducerClass(SelectTopKIPReducer.class);  
103.         job2.setMapOutputKeyClass(IPAndCount.class);  
104.         job2.setMapOutputValueClass(Text.class);  
105.         job2.setOutputKeyClass(IPAndCount.class);  
106.         job2.setOutputValueClass(Text.class);  
107.         job2.setNumReduceTasks(1);  
108.           
109.         JobControl jobControl = new JobControl("FindTopKIP");  
110.         ControlledJob cJob1 = new ControlledJob(conf);  
111.         cJob1.setJob(job);  
112.         ControlledJob cJob2 = new ControlledJob(conf2);  
113.         cJob2.setJob(job2);  
114.         jobControl.addJob(cJob1);  
115.         jobControl.addJob(cJob2);  
116.         cJob2.addDependingJob(cJob1);  
117.         jobControl.run();  
118.         return 0;  
119.     }  
120.       
121.     public static void main(String args[]) throws Exception{  
122.         int res = ToolRunner.run(new FindActiveIP(), args);  
123.         System.exit(res);  
124.     }  
125.   
126. }  

  大刀拿习惯了，从前的大刀就成了如今的绣花针，不是绣花针很差，只是用着不顺手。当你听着歌用java写着MapReduce，忽然有人在你耳边喊了一句：Pig~Pig~Pig~

  你很难不心动！程序员爱偷懒堪比女人爱逛街，都是为了快乐啊~

  下面是用Pig来处理上述的问题：

  

[plain]  view plain copy print ?

1. grunt> records = LOAD 'input/ipdata' AS (ip:chararray);                           
2. grunt> grouped_records = GROUP records BY ip;                                     
3. grunt> counted_records = FOREACH grouped_records GENERATE group, COUNT(records);  
4. grunt> sorted_records = ORDER counted_records BY $1 DESC;                         
5. grunt> topK = LIMIT sorted_records 10;                                                    
6. grunt> DUMP topK;    

你看，数数，没晕数一数，6行！仅仅6行就解决了。

行1：将文件装入

行2：按ip分组

行3：组内计数

行4：组间按ip访问计数排序

行5：选择前10个数据

行6：运行并输出。

虽然咱们的Pig方法实际上跑了3个job才完成任务，相比于java写的MapReduce多了一个job，但Pig显然更愉快些。

这是最后结果：

(192.168.0.1,1559)
(192.168.0.21,7)
(192.168.0.14,4)
(192.168.0.10,4)
(192.168.0.12,4)
(192.168.0.32,4)
(192.168.0.13,3)
(192.168.0.3,3)
(192.168.0.2,2)
(192.168.0.11,1)

这个算法对带宽的压力仍是比较大的，除了加一个Combiner以外，代码中还提到了另外一个小小的优化在进入第一个Reduce阶段的一个reduce task中的数据足以装入内存时，这是很容易解决的。这不是多好的优化，应当有更加好的优化方式能过滤更多的数据，不然。。。这不科学~

  二、寻找热门查询：搜索引擎会经过日志文件把用户每次检索使用的全部检索串都记录下来，每一个查询串的长度为1-255字节。

    假设目前有100亿个记录（这些查询串的重复度比较高，虽然总数是100亿，但若是除去重复后，不超过10亿个。一个查询串的重复度越高，说明查询它的用户越多，也就是越热门），请你统计最热门的10个查询串。[我仍是稍微修改了题目]

    看见这个题目，我以为我写下去会对不起July费了万千脑细胞辛苦的写做成果，尽管我心里十分但愿它跟上面那题同样，但这多么让人不甘心又不尽兴~

    如今暂且不考虑优化的问题：这个题目无非就是统计查询串的计数，而后排序，而后取出前10个。事实上，这个问题在不考虑细节上彻底能够用上面的pig脚原本处理。

不写了，以如今的水平继续写实在是不优美:
3题：有一个1G大小的一个文件，里面每一行是一个词，词的大小不超过16字节，内存限制大小是1M。返回频数最高的100个词。
4题：海量数据分布在100台电脑中，想个办法高效统计出这批数据的TOP10。
5题：有10个文件，每一个文件1G，每一个文件的每一行存放的都是用户的query，每一个文件的query均可能重复。要求你按照query的频度排序。
7题：怎么在海量数据中找出重复次数最多的一个？
8题：上千万或上亿数据（有重复），统计其中出现次数最多的前N个数据。
9题：一个文本文件，大约有一万行，每行一个词，要求统计出其中最频繁出现的前10个词。

1，2，3，4，5，7，8，9题思路基本一致，值得注意的是，有时候咱们彻底能够肯定咱们须要的数据的一些特征，好比上面的热门查询中热门串必定被查询超过1000次，那么咱们就可使用FILTER来进行过滤以减小处理的数据（从而减小对带宽的压力）[filted_records = FILTER grouped_records BY SIZE(records) > 1000;]。

6题： 给定a、b两个文件，各存放50亿个url，每一个url各占64字节，内存限制是4G，让你找出a、b文件共同的url？
见 [Hadoop]使用DistributedCache进行复制联结  以及  使用hadoop的datajoin包进行关系型join操做，你也能够参考Data-Intensive Text Processing with MapReduce看看原生态的join操做是怎么进行的。

[plain]  view plain copy print ?

1. grunt> A = LOAD 'input/url1' AS (url:chararray);  
2. grunt> B = LOAD 'input/url2' AS (url:chararray);  
3. grunt> grouped_A = GROUP A BY url;  
4. grunt> non_duplicated_A = FOREACH grouped_A GENERATE group;  --去重  
5. grunt> grouped_B = GROUP B BY url;  
6. grunt> non_duplicated_B = FOREACH grouped_B GENERATE group;  --B去重  
7. grunt> C = JOIN non_duplicated_B BY group, non_duplicated_A BY group;  --A 、B 内联结  
8. grunt> D = FOREACH C GENERATE $0;   //生成重复url  
9. grunt> DUMP D;  

10题： 1000万字符串，其中有些是重复的，须要把重复的所有去掉，保留没有重复的字符串。
使用pig：

[plain]  view plain copy print ?

1. grunt> records = LOAD 'input/retrived_strings' AS (str:chararray);  
2. grunt> grouped_records = GROUP records BY str;  
3. grunt> filted_records = FILTER grouped_records BY SIZE(records) <= 1;  
4. grunt> DUMP filted_records;  

   今日在CSDN看再次碰见July的这篇博文：教你如何迅速秒杀掉：99%的海量数据处理面试题。

   这篇文章我以前是拜读过的，今天闲来没事，就想拿来当作MapReduce的练习。

   MapReduce这把刀太大，刀大了问题就抵不住这刀锋了，事实上一开始我想着，这么多些题目，当是要花很多功夫的，但当我作完一题继续看下面的题目的时候，才发现这些题目在MapReduce模型下显得大同小异了，看来拿大刀的人是无论砍的是木头仍是人头的，而是直接抽象成柱形物而后抡起刀一刀就下去了。

     直入主题：

     一、海量日志数据，提取出某日访问百度次数最多的前K个IP。[稍微改变]

                   说明：每一次访问网页就在日志中记录1次访问者的IP，独占一行，一个小数据能够在这里下载。

       实在是想不出如何能在一个Job中解决这个问题，因此仍是把它拉扯成了两个Job来解决。
         Job1：将相同IP的记录合并，造成<ip,count>形式,其中count是对这个ip的计数。
         Job2：按count排序<ip,count>并选择前K个进行输出。
       这里我写了一个可序列化的类IPAndCount，若是稍微熟悉MapReduce或者看明白我以前写的 关系型MapReduce模式：选择、分组和组内排序 你就知道这是为了排序而准备的。MapReduce有一个“Shuffle and sort”，这个阶段是利用key来对tuple进行排序的，而排序时调用的即是key的compareTo()方法。事实上若是job1输出的数据两足够小，咱们彻底能够在内存中进行排序而利用MapReduce框架，这样就能够省下一个Reduce阶段，可是对于这个问题显然不行。
       IPAndCount很直白，就是包装了上述的<ip,count>。
   [java]  view plain copy print ?

   1. import java.io.DataInput;  
   2. import java.io.DataOutput;  
   3. import java.io.IOException;  
   4.   
   5. import org.apache.hadoop.io.IntWritable;  
   6. import org.apache.hadoop.io.Text;  
   7. import org.apache.hadoop.io.WritableComparable;  
   8.   
   9.   
   10. public class IPAndCount implements WritableComparable{  
   11.     Text ip;  
   12.     IntWritable count;  
   13.       
   14.     public IPAndCount(){  
   15.         this.ip = new Text("");  
   16.         this.count = new IntWritable(1);  
   17.     }  
   18.       
   19.     public IPAndCount(Text ip, IntWritable count){  
   20.         this.ip = ip;  
   21.         this.count = count;  
   22.     }  
   23.       
   24.     public IPAndCount(String ip, int count){  
   25.         this.ip = new Text(ip);  
   26.         this.count = new IntWritable(count);  
   27.     }  
   28.       
   29.     public void readFields(DataInput in) throws IOException {  
   30.         ip.readFields(in);  
   31.         count.readFields(in);  
   32.     }  
   33.   
   34.     public void write(DataOutput out) throws IOException {  
   35.         ip.write(out);  
   36.         count.write(out);  
   37.     }  
   38.   
   39.     public int compareTo(Object o) {  
   40.         return ((IPAndCount)o).count.compareTo(count) == 0?   
   41.                 ip.compareTo(((IPAndCount)o).ip):((IPAndCount)o).count.compareTo(count);//若是只比较count会丢失数据，应该是suffle阶段的问题  
   42.     }  
   43.       
   44.     public int hashCode(){  
   45.         return ip.hashCode();  
   46.     }  
   47.       
   48.     public boolean equals(Object o){  
   49.         if(!(o instanceof IPAndCount))  
   50.             return false;  
   51.         IPAndCount other = (IPAndCount)o;  
   52.         return ip.equals(other.ip) && count.equals(other.count);  
   53.     }  
   54.       
   55.     public String toString(){  
   56.         StringBuffer buf = new StringBuffer("[ip=");  
   57.         buf.append(ip.toString());  
   58.         buf.append(",count=");  
   59.         buf.append(count.toString());  
   60.         buf.append("]");  
   61.         return buf.toString();  
   62.     }  
   63.       
   64.     public Text getIp() {  
   65.         return ip;  
   66.     }  
   67.     public void setIp(Text ip) {  
   68.         this.ip = ip;  
   69.     }  
   70.     public IntWritable getCount() {  
   71.         return count;  
   72.     }  
   73.     public void setCount(IntWritable count) {  
   74.         this.count = count;  
   75.     }  
   76. }  

   
   

   下面对FindActiveIp进行说明：
   SumUpIpMapper和SumUpIPReducer事实上就是一个MapReduce中最基础的词频统计程序WordCount。你能够加一个Combiner来优化一下，我遗漏了。
   从配置中能够看见两个Job的配置：job 和job2。
   依赖关系是job -> job2，代码中使用了JobControl来解决做业间的依赖关系，JobControl.run()方法会在做业都运行完后才返回。
   Job2的输入路径是Job1的输出路径，从参数中能够看出这一点。
   Job1的输出在输出文件中的表现是：ip,count
   Job2再从文件中读入，使用的是KeyValueTextInputFormat，它对应的是TextOutputFormat，咱们能够从job1的配置中看出来。
   BeforeSortIPMapper从job1的输出中读取数据并包装成IPAndCount类型，以便MapReduce框架在“shuffle and sort”阶段利用它来排序。
   最后SelectTopKIPReducer选出前K个进行输出便可，在这里咱们设置最后的reduce只有一个reduce task，以使全部数据汇聚到一台机子上进行处理。
   [java]  view plain copy print ?

   1. import java.io.IOException;  
   2.   
   3. import org.apache.hadoop.conf.Configuration;  
   4. import org.apache.hadoop.conf.Configured;  
   5. import org.apache.hadoop.fs.Path;  
   6. import org.apache.hadoop.io.IntWritable;  
   7. import org.apache.hadoop.io.LongWritable;  
   8. import org.apache.hadoop.io.Text;  
   9. import org.apache.hadoop.mapred.lib.ChainMapper;  
   10. import org.apache.hadoop.mapreduce.Job;  
   11. import org.apache.hadoop.mapreduce.Mapper;  
   12. import org.apache.hadoop.mapreduce.Reducer;  
   13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
   14. import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;  
   15. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
   16. import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;  
   17. import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;  
   18. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
   19. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;  
   20. import org.apache.hadoop.util.Tool;  
   21. import org.apache.hadoop.util.ToolRunner;  
   22.   
   23.   
   24. public class FindActiveIP extends Configured implements Tool{  
   25.       
   26.     public static class SumUpIPMapper extends Mapper<LongWritable,Text,Text,IntWritable>{  
   27.         IntWritable  one = new IntWritable(1);  
   28.         public void map(LongWritable key, Text value, Context context)   
   29.             throws IOException,InterruptedException{  
   30.             context.write(value, one);  
   31.         }  
   32.     }  
   33.   
   34.     public static class SumUpIPReducer extends Reducer<Text,IntWritable,Text,IntWritable>{  
   35.         //这里能够选择前k个进行输出以优化  
   36.         public void reduce(Text key, Iterable<IntWritable> values, Context context)  
   37.             throws IOException, InterruptedException{  
   38.             int sum = 0;  
   39.             for(IntWritable v : values){  
   40.                 sum += v.get();  
   41.             }  
   42.             context.write(key, new IntWritable(sum));  
   43.         }  
   44.     }  
   45.       
   46.       
   47.     public static class BeforeSortIPMapper extends Mapper<Text,Text,IPAndCount,Text>{  
   48.         public void map(Text key, Text value, Context context)  
   49.             throws IOException,InterruptedException{  
   50.             IPAndCount tmp = new IPAndCount(key,new IntWritable(Integer.valueOf(value.toString())));  
   51.             System.out.println(tmp);  
   52.             context.write(tmp,new Text());  
   53.         }  
   54.     }  
   55.       
   56.       
   57.     //set num of this reducer to one  
   58.     public static class SelectTopKIPReducer extends Reducer<IPAndCount,Text,IPAndCount,Text>{  
   59.         int counter = 0;  
   60.         int K = 10;  
   61.         public void reduce(IPAndCount key, Iterable<Text> values, Context context)  
   62.             throws IOException, InterruptedException{  
   63.             System.out.println(key);  
   64.             if(counter < K){  
   65.                 context.write(key, null);  
   66.                   
   67.                 counter++;  
   68.             }  
   69.               
   70.         }  
   71.     }  
   72.     public int run(String[] args) throws Exception {  
   73.         Configuration conf = new Configuration();  
   74.         Job job = new Job(conf,"SumUpIP");  
   75.         job.setJarByClass(FindActiveIP.class);  
   76.         job.setInputFormatClass(TextInputFormat.class);  
   77.         job.setOutputFormatClass(TextOutputFormat.class);  
   78.         job.getConfiguration().set("mapred.textoutputformat.separator", ",");  
   79.         Path in = new Path(args[0]);  
   80.         Path out = new Path(args[1]);  
   81.         FileInputFormat.setInputPaths(job, in);  
   82.         FileOutputFormat.setOutputPath(job, out);  
   83.         job.setMapperClass(SumUpIPMapper.class);  
   84.         job.setReducerClass(SumUpIPReducer.class);  
   85.         job.setMapOutputKeyClass(Text.class);  
   86.         job.setMapOutputValueClass(IntWritable.class);  
   87.         job.setOutputKeyClass(Text.class);  
   88.         job.setOutputValueClass(IntWritable.class);  
   89.         job.setNumReduceTasks(7);  
   90.           
   91.         Configuration conf2 = new Configuration();  
   92.         Job job2 = new Job(conf2,"SortAndFindTopK");  
   93.         job2.setJarByClass(FindActiveIP.class);  
   94.         job2.setInputFormatClass(KeyValueTextInputFormat.class);  
   95.         job2.getConfiguration().set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ",");  
   96.         job2.setOutputFormatClass(TextOutputFormat.class);  
   97.         Path in2 = new Path(args[1]);  
   98.         Path out2 = new Path(args[2]);  
   99.         FileInputFormat.setInputPaths(job2,in2);  
   100.         FileOutputFormat.setOutputPath(job2, out2);  
   101.         job2.setMapperClass(BeforeSortIPMapper.class);  
   102.         job2.setReducerClass(SelectTopKIPReducer.class);  
   103.         job2.setMapOutputKeyClass(IPAndCount.class);  
   104.         job2.setMapOutputValueClass(Text.class);  
   105.         job2.setOutputKeyClass(IPAndCount.class);  
   106.         job2.setOutputValueClass(Text.class);  
   107.         job2.setNumReduceTasks(1);  
   108.           
   109.         JobControl jobControl = new JobControl("FindTopKIP");  
   110.         ControlledJob cJob1 = new ControlledJob(conf);  
   111.         cJob1.setJob(job);  
   112.         ControlledJob cJob2 = new ControlledJob(conf2);  
   113.         cJob2.setJob(job2);  
   114.         jobControl.addJob(cJob1);  
   115.         jobControl.addJob(cJob2);  
   116.         cJob2.addDependingJob(cJob1);  
   117.         jobControl.run();  
   118.         return 0;  
   119.     }  
   120.       
   121.     public static void main(String args[]) throws Exception{  
   122.         int res = ToolRunner.run(new FindActiveIP(), args);  
   123.         System.exit(res);  
   124.     }  
   125.   
   126. }  

   
     大刀拿习惯了，从前的大刀就成了如今的绣花针，不是绣花针很差，只是用着不顺手。当你听着歌用java写着MapReduce，忽然有人在你耳边喊了一句：Pig~Pig~Pig~
     你很难不心动！程序员爱偷懒堪比女人爱逛街，都是为了快乐啊~
     下面是用Pig来处理上述的问题：
     

   [plain]  view plain copy print ?

   1. grunt> records = LOAD 'input/ipdata' AS (ip:chararray);                           
   2. grunt> grouped_records = GROUP records BY ip;                                     
   3. grunt> counted_records = FOREACH grouped_records GENERATE group, COUNT(records);  
   4. grunt> sorted_records = ORDER counted_records BY $1 DESC;                         
   5. grunt> topK = LIMIT sorted_records 10;                                                    
   6. grunt> DUMP topK;    

   
   
   你看，数数，没晕数一数，6行！仅仅6行就解决了。
   行1：将文件装入
   行2：按ip分组
   行3：组内计数
   行4：组间按ip访问计数排序
   行5：选择前10个数据
   行6：运行并输出。
   
   
   虽然咱们的Pig方法实际上跑了3个job才完成任务，相比于java写的MapReduce多了一个job，但Pig显然更愉快些。
   
   
   这是最后结果：
   (192.168.0.1,1559)
   (192.168.0.21,7)
   (192.168.0.14,4)
   (192.168.0.10,4)
   (192.168.0.12,4)
   (192.168.0.32,4)
   (192.168.0.13,3)
   (192.168.0.3,3)
   (192.168.0.2,2)
   (192.168.0.11,1)
   
   
   
   这个算法对带宽的压力仍是比较大的，除了加一个Combiner以外，代码中还提到了另外一个小小的优化在进入第一个Reduce阶段的一个reduce task中的数据足以装入内存时，这是很容易解决的。这不是多好的优化，应当有更加好的优化方式能过滤更多的数据，不然。。。这不科学~
   
   
     二、寻找热门查询：搜索引擎会经过日志文件把用户每次检索使用的全部检索串都记录下来，每一个查询串的长度为1-255字节。
       假设目前有100亿个记录（这些查询串的重复度比较高，虽然总数是100亿，但若是除去重复后，不超过10亿个。一个查询串的重复度越高，说明查询它的用户越多，也就是越热门），请你统计最热门的10个查询串。[我仍是稍微修改了题目]
       看见这个题目，我以为我写下去会对不起July费了万千脑细胞辛苦的写做成果，尽管我心里十分但愿它跟上面那题同样，但这多么让人不甘心又不尽兴~
       如今暂且不考虑优化的问题：这个题目无非就是统计查询串的计数，而后排序，而后取出前10个。事实上，这个问题在不考虑细节上彻底能够用上面的pig脚原本处理。
   
   不写了，以如今的水平继续写实在是不优美:
   3题：有一个1G大小的一个文件，里面每一行是一个词，词的大小不超过16字节，内存限制大小是1M。返回频数最高的100个词。
   4题：海量数据分布在100台电脑中，想个办法高效统计出这批数据的TOP10。
   5题：有10个文件，每一个文件1G，每一个文件的每一行存放的都是用户的query，每一个文件的query均可能重复。要求你按照query的频度排序。
   7题：怎么在海量数据中找出重复次数最多的一个？
   8题：上千万或上亿数据（有重复），统计其中出现次数最多的前N个数据。
   9题：一个文本文件，大约有一万行，每行一个词，要求统计出其中最频繁出现的前10个词。
   
   1，2，3，4，5，7，8，9题思路基本一致，值得注意的是，有时候咱们彻底能够肯定咱们须要的数据的一些特征，好比上面的热门查询中热门串必定被查询超过1000次，那么咱们就可使用FILTER来进行过滤以减小处理的数据（从而减小对带宽的压力）[filted_records = FILTER grouped_records BY SIZE(records) > 1000;]。
   
   6题： 给定a、b两个文件，各存放50亿个url，每一个url各占64字节，内存限制是4G，让你找出a、b文件共同的url？
   见 [Hadoop]使用DistributedCache进行复制联结  以及  使用hadoop的datajoin包进行关系型join操做，你也能够参考Data-Intensive Text Processing with MapReduce看看原生态的join操做是怎么进行的。
   

   [plain]  view plain copy print ?

   1. grunt> A = LOAD 'input/url1' AS (url:chararray);  
   2. grunt> B = LOAD 'input/url2' AS (url:chararray);  
   3. grunt> grouped_A = GROUP A BY url;  
   4. grunt> non_duplicated_A = FOREACH grouped_A GENERATE group;  --去重  
   5. grunt> grouped_B = GROUP B BY url;  
   6. grunt> non_duplicated_B = FOREACH grouped_B GENERATE group;  --B去重  
   7. grunt> C = JOIN non_duplicated_B BY group, non_duplicated_A BY group;  --A 、B 内联结  
   8. grunt> D = FOREACH C GENERATE $0;   //生成重复url  
   9. grunt> DUMP D;  

   
   10题： 1000万字符串，其中有些是重复的，须要把重复的所有去掉，保留没有重复的字符串。
   使用pig：
   

   [plain]  view plain copy print ?

   1. grunt> records = LOAD 'input/retrived_strings' AS (str:chararray);  
   2. grunt> grouped_records = GROUP records BY str;  
   3. grunt> filted_records = FILTER grouped_records BY SIZE(records) <= 1;  
   4. grunt> DUMP filted_records;