Hadoop基础-Map端链式编程之MapReduce统计TopN示例

 1 /*
 2 @author :yinzhengjie
 3 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
 4 EMAIL:y1053419035@qq.com
 5 */
 6 package cn.org.yinzhengjie.mrchain;
 7 
 8 import org.apache.hadoop.io.WritableComparable;
 9 
10 import java.io.DataInput;
11 import java.io.DataOutput;
12 import java.io.IOException;
13 
14 public class CompKey implements WritableComparable<CompKey> {
15 
16     private String year;
17     private int temp;
18     /**
19      * 重写CompKey对年份和气温排序
20      */
21     public int compareTo(CompKey o) {
22         if(this.getYear().equals(o.getYear())){
23             return o.getTemp() - this.getTemp();
24         }
25         return this.getYear().compareTo(o.getYear());
26 
27     }
28 
29     public void write(DataOutput out) throws IOException {
30         out.writeUTF(year);
31         out.writeInt(temp);
32 
33     }
34 
35     public void readFields(DataInput in) throws IOException {
36         year = in.readUTF();
37         temp = in.readInt();
38 
39     }
40 
41     public String getYear() {
42         return year;
43     }
44 
45     public void setYear(String year) {
46         this.year = year;
47     }
48 
49     public int getTemp() {
50         return temp;
51     }
52 
53     public void setTemp(int temp) {
54         this.temp = temp;
55     }
56 
57     @Override
58     public String toString() {
59         return year + '\t' +temp ;
60     }
61 }

 1 /*
 2 @author :yinzhengjie
 3 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
 4 EMAIL:y1053419035@qq.com
 5 */
 6 package cn.org.yinzhengjie.mrchain;
 7 
 8 import org.apache.hadoop.io.WritableComparable;
 9 import org.apache.hadoop.io.WritableComparator;
10 
11 public class MyGroupComparator extends WritableComparator {
12 
13     public MyGroupComparator() {
14         super(CompKey.class,true);
15     }
16 
17     public int compare(WritableComparable a, WritableComparable b) {
18         CompKey ck1 = (CompKey) a;
19         CompKey ck2 = (CompKey) b;
20         return ck1.getYear().compareTo(ck2.getYear());
21     }
22 }

 1 /*
 2 @author :yinzhengjie
 3 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
 4 EMAIL:y1053419035@qq.com
 5 */
 6 package cn.org.yinzhengjie.mrchain;
 7 
 8 import org.apache.hadoop.io.IntWritable;
 9 import org.apache.hadoop.io.LongWritable;
10 import org.apache.hadoop.io.Text;
11 import org.apache.hadoop.mapreduce.Mapper;
12 
13 import java.io.IOException;
14 
15 public class ChainMapper1 extends Mapper<LongWritable, Text, Text, IntWritable> {
16 
17     @Override
18     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
19 
20         String line = value.toString();
21 
22         //得到年份
23         String year = line.substring(15, 19);
24 
25         //得到气温
26         int temp = Integer.parseInt(line.substring(87, 92));
27 
28         context.write(new Text(year), new IntWritable(temp));
29 
30     }
31 }

 1 /*
 2 @author :yinzhengjie
 3 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
 4 EMAIL:y1053419035@qq.com
 5 */
 6 package cn.org.yinzhengjie.mrchain;
 7 
 8 import org.apache.hadoop.io.IntWritable;
 9 import org.apache.hadoop.io.NullWritable;
10 import org.apache.hadoop.io.Text;
11 import org.apache.hadoop.mapreduce.Mapper;
12 
13 import java.io.IOException;
14 
15 public class ChainMapper2 extends Mapper<Text,IntWritable,CompKey,NullWritable> {
16 
17     @Override
18     protected void map(Text key, IntWritable value, Context context) throws IOException, InterruptedException {
19 
20         int i = value.get();
21 
22         if( i != 9999){
23             CompKey ck = new CompKey();
24             ck.setYear(key.toString());
25             ck.setTemp(i);
26             context.write(ck,NullWritable.get());
27         }
28     }
29 }

 1 /*
 2 @author :yinzhengjie
 3 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
 4 EMAIL:y1053419035@qq.com
 5 */
 6 package cn.org.yinzhengjie.mrchain;
 7 
 8 import org.apache.hadoop.io.IntWritable;
 9 import org.apache.hadoop.io.NullWritable;
10 import org.apache.hadoop.io.Text;
11 import org.apache.hadoop.mapreduce.Reducer;
12 
13 import java.io.IOException;
14 import java.util.Iterator;
15 
16 public class ChainReducer1 extends Reducer<CompKey, NullWritable, Text, IntWritable> {
17 
18     //由于分组对比器设定，相同的year放在一个分组,因此，在一个reduce循环中，得到的数据均为同一年份的数据
19     protected void reduce(CompKey key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
20         String year = key.getYear();
21         Iterator<NullWritable> it = values.iterator();
22         int i = 0;
23         while (it.hasNext()){
24             System.out.println(key.toString());
25             int temp = key.getTemp();
26             context.write(new Text(year), new IntWritable(temp));
27             it.next();
28             i++;
29             if(i >= 10){
30                 break;
31             }
32         }
33     }
34 }

 1 /*
 2 @author :yinzhengjie
 3 Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
 4 EMAIL:y1053419035@qq.com
 5 */
 6 package cn.org.yinzhengjie.mrchain;
 7 
 8 import org.apache.hadoop.io.IntWritable;
 9 import org.apache.hadoop.io.Text;
10 import org.apache.hadoop.mapreduce.Mapper;
11 
12 import java.io.IOException;
13 
14 public class ChainReducer2 extends Mapper<Text, IntWritable, Text,IntWritable> {
15     protected void map(Text key, IntWritable value, Context context) throws IOException, InterruptedException {
16         int temp = value.get();
17         //取得奇数气温
18         if( temp % 2 == 1 ){
19             context.write(key, new IntWritable(temp));
20         }
21 
22     }
23 }