MapReduce实例分析

拼搏现实的明天。 2022-03-14 12:26 268阅读 0赞

0.MaxValue:要求输出cite75\_99.txt中最大的CITED值：

要点：

1.Mapper只输出它所处理的数据中的最大值。（重写cleanup()函数）

2.设置Reducer数目为一个 -D mapred.reduce.tasks=1，同时也只输出所处理的最大值。（重写cleanup()函数）

3.cleanup()函数：在任务结束时执行一次。详见API。

代码如下：

**\[java\]** [ view plain][view plain] [copy][view plain] [print][view plain] [?][view plain]

1.  /\* 
2.   \* MaxValues 
3.   \* 函数作用：输出Patent中最大数值 
4.   \* Author: jokes000 
5.   \* Date: 2011-12-15 
6.   \*/  
7.    
8.  import java.io.IOException;  
9.    
10. import org.apache.hadoop.conf.Configuration;  
11. import org.apache.hadoop.conf.Configured;  
12. import org.apache.hadoop.fs.Path;  
13. import org.apache.hadoop.io.IntWritable;  
14. import org.apache.hadoop.io.LongWritable;  
15. import org.apache.hadoop.io.Text;  
16. import org.apache.hadoop.mapreduce.Job;  
17. import org.apache.hadoop.mapreduce.Mapper;  
18. import org.apache.hadoop.mapreduce.Reducer;  
19. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
20. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
21. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
22. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;  
23. import org.apache.hadoop.util.Tool;  
24. import org.apache.hadoop.util.ToolRunner;  
25.   
26.   
27. public class MaxValue extends Configured implements Tool \{  
28.   
29.     public static class MapClass extends Mapper<LongWritable,Text,Text,Text> \{  
30.           
31.         int max = 0;  
32.         // Map Method   
33.         public void map(LongWritable key, Text value, Context context)\{  
34.             String\[\] citation = value.toString().split(",", 2);  
35.             try \{  
36.                 int tmp = Integer.parseInt(citation\[0\]);  
37.                 if( tmp > max ) max = tmp;  
38.             \} catch(NumberFormatException e)\{  
39.                 // do nothing.   
40.             \}  
41.             //context.write(new Text(citation\[0\]), new Text(citation\[0\]));   
42.         \}  
43.           
44.         @Override  
45.         protected void cleanup(Context context) throws IOException, InterruptedException \{  
46.             context.write(new Text(max+""), new Text(max+""));  
47.         \}  
48.     \}  
49.       
50.     public static class Reduce extends Reducer<Text,Text,Text,IntWritable> \{  
51.           
52.         int max = 0;  
53.         // Reduce Method    
54.         public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException \{  
55.             //IntWritable\[\] top = new IntWritable\[10\];   
56.             for(Text value : values) \{  
57.                 try \{  
58.                     int tmp = Integer.parseInt(value.toString());  
59.                     if( tmp > max ) max = tmp;  
60.                 \} catch(NumberFormatException e) \{  
61.                     // do nothing.   
62.                 \}  
63.             \}  
64.             //context.write(new Text("Max"), new IntWritable(max));   
65.         \}  
66.           
67.         @Override  
68.         protected void cleanup(Context context) throws IOException, InterruptedException \{  
69.             context.write(new Text("Max"), new IntWritable(max));  
70.         \}  
71.     \}  
72.       
73.     @Override  
74.     public int run(String\[\] arg0) throws Exception \{  
75.           
76.         Job job = new Job();  
77.         job.setJarByClass(MaxValue.class);  
78.           
79.         FileInputFormat.addInputPath(job, new Path(arg0\[0\]));  
80.         FileOutputFormat.setOutputPath(job, new Path(arg0\[1\]));  
81.           
82.         job.setMapperClass(MapClass.class);  
83.         job.setReducerClass(Reduce.class);  
84.         job.setInputFormatClass(TextInputFormat.class);  
85.         job.setOutputFormatClass(TextOutputFormat.class);  
86.         job.setOutputKeyClass(Text.class);  
87.         job.setOutputValueClass(Text.class);  
88.           
89.         job.waitForCompletion(true);  
90.           
91.         return 0;  
92.     \}  
93.       
94.     public static void main(String\[\] args) throws Exception \{  
95.         int res = ToolRunner.run(new Configuration(), new MaxValue(), args);  
96.         System.exit(res);  
97.     \}  
98.   
99. \}

/*
     * MaxValues
     * 函数作用：输出Patent中最大数值
     * Author: jokes000
     * Date: 2011-12-15
     */
    
    import java.io.IOException;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    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.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    
    public class MaxValue extends Configured implements Tool {
    
    	public static class MapClass extends Mapper<LongWritable,Text,Text,Text> {
    		
    		int max = 0;
    		// Map Method
    		public void map(LongWritable key, Text value, Context context){
    			String[] citation = value.toString().split(",", 2);
    			try {
    				int tmp = Integer.parseInt(citation[0]);
    				if( tmp > max ) max = tmp;
    			} catch(NumberFormatException e){
    				// do nothing.
    			}
    			//context.write(new Text(citation[0]), new Text(citation[0]));
    		}
    		
    		@Override
    		protected void cleanup(Context context) throws IOException, InterruptedException {
    			context.write(new Text(max+""), new Text(max+""));
    		}
    	}
    	
    	public static class Reduce extends Reducer<Text,Text,Text,IntWritable> {
    		
    		int max = 0;
    		// Reduce Method
    		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
    			//IntWritable[] top = new IntWritable[10];
    			for(Text value : values) {
    				try {
    					int tmp = Integer.parseInt(value.toString());
    					if( tmp > max ) max = tmp;
    				} catch(NumberFormatException e) {
    					// do nothing.
    				}
    			}
    			//context.write(new Text("Max"), new IntWritable(max));
    		}
    		
    		@Override
    		protected void cleanup(Context context) throws IOException, InterruptedException {
    			context.write(new Text("Max"), new IntWritable(max));
    		}
    	}
    	
    	@Override
    	public int run(String[] arg0) throws Exception {
    		
    		Job job = new Job();
    		job.setJarByClass(MaxValue.class);
    		
    		FileInputFormat.addInputPath(job, new Path(arg0[0]));
    		FileOutputFormat.setOutputPath(job, new Path(arg0[1]));
    		
    		job.setMapperClass(MapClass.class);
    		job.setReducerClass(Reduce.class);
    		job.setInputFormatClass(TextInputFormat.class);
    		job.setOutputFormatClass(TextOutputFormat.class);
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(Text.class);
    		
    		job.waitForCompletion(true);
    		
    		return 0;
    	}
    	
    	public static void main(String[] args) throws Exception {
    		int res = ToolRunner.run(new Configuration(), new MaxValue(), args);
    		System.exit(res);
    	}
    
    }

1.Top  K Values: 要求输出adapt63\_99.txt中的第9列CLAIMS值的最大的K个值：

要点：

1.Mapper值输出它所处理的数据中的最大的K个值。(重写 cleanup()函数)

2.设置Reducer数目为1 -D mapred.reduce.tasks=1，同时对Mapper中输出进行排序，输出最大的K个值(重写 cleanup()函数)

代码如下：

**\[java\]** [ view plain][view plain] [copy][view plain] [print][view plain] [?][view plain]

1.  /\* 
2.   \* TopKValues 
3.   \* 函数作用：输出CLAIMS中最大的几个数值 
4.   \* Author: jokes000 
5.   \* Date: 2011-12-15 
6.   \*/  
7.    
8.  import java.io.IOException;  
9.  import java.util.Arrays;  
10.   
11. import org.apache.hadoop.conf.Configuration;  
12. import org.apache.hadoop.conf.Configured;  
13. import org.apache.hadoop.fs.Path;  
14. import org.apache.hadoop.io.IntWritable;  
15. import org.apache.hadoop.io.LongWritable;  
16. import org.apache.hadoop.io.Text;  
17. import org.apache.hadoop.mapreduce.Job;  
18. import org.apache.hadoop.mapreduce.Mapper;  
19. import org.apache.hadoop.mapreduce.Reducer;  
20. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
21. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
22. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
23. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;  
24. import org.apache.hadoop.util.Tool;  
25. import org.apache.hadoop.util.ToolRunner;  
26.   
27.   
28. public class TopKValues extends Configured implements Tool \{  
29.   
30.     public static class MapClass extends Mapper<LongWritable,Text,Text,IntWritable> \{  
31.         // 全局变量   
32.         int len;    // K值   
33.         int\[\] top;  // 用于保存的数组   
34.           
35.         // Map Method   
36.         public void map(LongWritable key, Text value, Context context) \{  
37.             String\[\] fields = value.toString().split(",",-20);  
38.             try \{  
39.                 int claims = Integer.parseInt(fields\[8\]);  
40.                 add(claims);  
41.             \} catch(NumberFormatException e) \{  
42.                 // do nothing..   
43.             \}  
44.         \}  
45.           
46.         private void add(int value) \{  
47.             top\[0\] = value;  
48.             Arrays.sort(top);  
49.         \}  
50.           
51.         @Override  
52.         protected void setup(Context context) \{  
53.             // 获取设置的"K"值，若没有K值，则设置该值为10   
54.             len = context.getConfiguration().getInt("K", 10);  
55.             top = new int\[len+1\];  
56.         \}  
57.           
58.         @Override  
59.         protected void cleanup(Context context) throws IOException, InterruptedException \{  
60.             for( int i = 1; i <= len; ++ i ) \{  
61.                 context.write(new Text(top\[i\]+""), new IntWritable(top\[i\]));  
62.             \}  
63.         \}  
64.     \}  
65.       
66.     public static class Reduce extends Reducer<Text,IntWritable,Text,IntWritable> \{  
67.         int\[\] top;  
68.         int len;  
69.           
70.         @Override  
71.         protected void setup(Context context) \{  
72.             len = context.getConfiguration().getInt("K", 10);  
73.             top = new int\[len+1\];  
74.         \}  
75.           
76.         private void add(int value) \{  
77.             top\[0\] = value;  
78.             Arrays.sort(top);  
79.         \}  
80.           
81.         // Reduce Method    
82.         public void reduce(Text key, Iterable<IntWritable> values, Context context) \{  
83.             for(IntWritable value : values) \{  
84.                 add(value.get());  
85.             \}  
86.         \}  
87.           
88.         @Override  
89.         protected void cleanup(Context context) throws IOException, InterruptedException \{  
90.             for( int i = len; i > 0; -- i ) \{  
91.                 context.write(new Text("No."\+(len-i+1)), new IntWritable(top\[i\]));  
92.             \}  
93.         \}  
94.     \}  
95.       
96.     @Override  
97.     public int run(String\[\] arg0) throws Exception \{  
98.           
99.         Job job = new Job();  
100.         job.setJarByClass(TopKValues.class);  
101.           
102.         FileInputFormat.addInputPath(job, new Path(arg0\[0\]));  
103.         FileOutputFormat.setOutputPath(job, new Path(arg0\[1\]));  
104.           
105.         try\{  
106.             int K = Integer.parseInt(arg0\[2\]);  
107.             getConf().setInt("K", K);  
108.         \} catch(NumberFormatException e) \{  
109.             // do nothing..    
110.             getConf().setInt("K", 20);  
111.         \}  
112.           
113.         job.setMapperClass(MapClass.class);  
114.         job.setReducerClass(Reduce.class);  
115.         job.setInputFormatClass(TextInputFormat.class);  
116.         job.setOutputFormatClass(TextOutputFormat.class);  
117.         job.setOutputKeyClass(Text.class);  
118.         job.setOutputValueClass(IntWritable.class);  
119.           
120.         job.waitForCompletion(true);  
121.           
122.         return 0;  
123.     \}  
124.       
125.     public static void main(String\[\] args) throws Exception \{  
126.         int res = ToolRunner.run(new Configuration(), new TopKValues(), args);  
127.         System.exit(res);  
128.     \}  
129.   
130. \}

[view plain]: http://blog.csdn.net/jokes000/article/details/7075344#