试着写写Map/Reduce程序

手头有大量网址，以文本格式存储，一行一个url，像这样：

1. http://beijing.baixing.com/zhengzu/a232119437.html 
2. http://mall.cnki.net/magazine/Article/JCTD199507000.htm 
3. http://meishi.qq.com/shops/2847395840683387518 
4. http://beijing.baixing.com/zhengzu/a233512411.html 
5. http://meishi.qq.com/shops/2710663397051226108 

现在想按域名整理成下面这种格式：

1. beijing.baixing.com 
2. http://beijing.baixing.com/zhengzu/a232119437.html 
3. http://beijing.baixing.com/zhengzu/a233512411.html 
4.  
5. mall.cnki.net 
6. http://mall.cnki.net/magazine/Article/JCTD199507000.htm 
7.  
8. meishi.qq.com 
9. http://meishi.qq.com/shops/2847395840683387518 
10. http://meishi.qq.com/shops/2710663397051226108 

正好想到了用hadoop试试，于是试着写了一个Map/Reduce程序，太简单了，只记录不解释。

1. package com.rs; 
2.  
3. import java.io.*; 
4. import java.util.*; 
5.  
6. import org.apache.hadoop.fs.*; 
7. import org.apache.hadoop.io.*; 
8. import org.apache.hadoop.mapred.*; 
9.  
10. public class Url { 
11.  
12.     /** 
13.      * 实现map 
14.      */ 
15.     @SuppressWarnings("deprecation") 
16.     public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text>{ 
17.  
18.         /** 
19.          * map方法，提取每个url的域名 
20.          */ 
21.         public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter report) throws IOException{ 
22.          
23.             //value是一行的内容，output是map的结果输出 
24.             String line = value.toString(); 
25.             int pos = line.indexOf("/", 7); 
26.             String domain = line.substring(7, pos); 
27.              
28.             //map的结果为(domain, url) 
29.             output.collect(new Text(domain), value); 
30.         } 
31.     } 
32.      
33.     /** 
34.      * 实现reduce 
35.      */ 
36.     @SuppressWarnings("deprecation") 
37.     public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text>{ 
38.          
39.         /** 
40.          * reduce方法，将map的结果聚合 
41.          */ 
42.         public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter report) throws IOException{ 
43.              
44.             //key就是map后的那个domain，values是map后的一个个url集合，output是reduce后的结果输出 
45.             //把每一个url用\n连起来 
46.             StringBuilder urls = new StringBuilder(); 
47.             urls.append("\n"); 
48.             while(values.hasNext()){ 
49.                 urls.append(values.next().toString()); 
50.                 urls.append("\n"); 
51.             } 
52.              
53.             //reduce的结果为(domain, urls) 
54.             output.collect(key, new Text(urls.toString())); 
55.         } 
56.     } 
57.      
58.     @SuppressWarnings("deprecation") 
59.     public static void main(String[] args) throws Exception{ 
60.          
61.         JobConf conf = new JobConf(Url.class); 
62.         conf.setJobName("urlanalyse");          //任务名称 
63.          
64.         conf.setOutputKeyClass(Text.class); 
65.         conf.setOutputValueClass(Text.class); 
66.          
67.         //设置map与reduce的类 
68.         conf.setMapperClass(Map.class); 
69.         conf.setReducerClass(Reduce.class); 
70.          
71.         //输入输出均为文本格式，所以用TextInputFormat和TextOutputFormat 
72.         //可以换其它的比如DBOutputFormat输出到数据库，也可以自定义 
73.         conf.setInputFormat(TextInputFormat.class); 
74.         conf.setOutputFormat(TextOutputFormat.class); 
75.          
76.         //指定输入输出(由命令行参数控制) 
77.         FileInputFormat.setInputPaths(conf, new Path(args[0])); 
78.         FileOutputFormat.setOutputPath(conf, new Path(args[1])); 
79.          
80.         //执行任务 
81.         JobClient.runJob(conf); 
82.     } 
83. } 

导出成jar文件，放到hadoop目录下，用命令行跑一下：

1. #在dfs上清空input与output目录 
2. hadoop@rs:/usr/local/hadoop$ bin/hadoop fs -rmr input 
3. hadoop@rs:/usr/local/hadoop$ bin/hadoop fs -rmr output 
4.  
5. #将url.txt放到dfs上的input目录下 
6. hadoop@rs:/usr/local/hadoop$ bin/hadoop fs -put url.txt input/url.txt 
7.  
8. #对dfs上的input目录中的文件执行map/reduce，输出结果放到output中 
9. hadoop@rs:/usr/local/hadoop$ bin/hadoop jar url.jar input output 
10.  
11. #查看一下dfs上output目录下的文件内容 
12. hadoop@rs:/usr/local/hadoop$ bin/hadoop fs -cat output/part* 

搞定收工。

PS：单机伪分布模式下，测了一下500万条url跑了110秒，有点慢。等正式数据来了上集群环境试试。
PPS：环境为Ubuntu 10.10 + Sun JDK 1.6.38 + Hadoop 0.20.2

     本文转自 BoyTNT 51CTO博客，原文链接：http://blog.51cto.com/boytnt/1089195，如需转载请自行联系原作者