Apache Nutch 1.3 学习笔记三（Inject CrawlDB Reader）

上次我们分析了Inject的整个流程，其中说到了Inject的输出格式是MapSequenceFileOutputFormat，这个格式可以使用一个叫CrawlDbReader的工具来读取和分析。下面我们就来分析一下这个工具有哪些用。

1. CrawlDbReader工具的使用方法

   在命令行中运行bin/nutch readdb后就可以看到其帮助，实际上这个shell方法调用的正是CrawlDbReader的main方法，这个工具有下面几种使用方法：

   * bin/nutch <crawldb> -stats -sort

     这个方法是在终端中打印所有crawldb的统计信息，加上sort后，会输出所有host所对应的url的个数，一般格式如下

 

1. lemo@debian:~/Workspace/java/Apache/Nutch/nutch-1.3$ bin/nutch readdb db/crawldb/ -stats -sort  
2.     CrawlDb statistics start: db/crawldb/  
3.     Statistics for CrawlDb: db/crawldb/  
4.     TOTAL urls: 5  
5.     retry 0:    5  
6.     min score:  0.045  
7.     avg score:  0.09955  
8.     max score:  1.136  
9.     status 1 (db_unfetched):  4   
10.        baike.baidu.com :    1  
11.        hi.baidu.com :   2  
12.        home.baidu.com : 1  
13.        image.baidu.com :    1  
14.     status 2 (db_fetched):  1  
15.             www.baidu.com : 1  
16.     CrawlDb statistics: done  

   * bin/nutch <crawldb> -dump <out_dir> [-format normal|csv]

     这个方法主要是把CrawlDb中的数据转成一般的文本格式或者是csv格式，输出到out_dir目录中

   * bin/nutch <crawldb> -url <url>

     这个方法主要是打印出特定url的信息

   * bin/nutch <crawldb> -topN nnn <out_dir> [<min>]

     这个方法主要是输出前topN个分数大于min的url，输出到out_dir中，默认这个min为0.0,本地的输出如下

 

1. lemo@debian:~/Workspace/java/Apache/Nutch/nutch-1.3$ bin/nutch readdb db/crawldb/ -topN 3 out  
2.     CrawlDb topN: starting (topN=3, min=0.0)  
3.     CrawlDb db: db/crawldb/  
4.     CrawlDb topN: collecting topN scores.  
5.     CrawlDb topN: done  
6. lemo@debian:~/Workspace/java/Apache/Nutch/nutch-1.3$ cat out/part-00000   
7.     1.1363636   http://www.baidu.com/  
8.     0.045454547 http://tieba.baidu.com/  
9.     0.045454547 http://baike.baidu.com/  

2. 下面简单分析一下上面四个流程的源代码

   * bin/nutch <crawldb> -stats -sort

     这个命令是调用CrawlDbReader中的processStatJob(String CrawlDb,Configuration config,boolean sort)来解决掉的
下面用一个MP任务对CrawlDb数据库进行统计，主要代码如下
 

 

1. JobConf job = new NutchJob(config);   // 生成一个Job的配置对象  
2.      job.setJobName("stats " + crawlDb);  
3.      job.setBoolean("db.reader.stats.sort", sort);   // 配置是否要进行sort操作，这个标记会在CrawlDbStatMapper中用到  
4.     
5.     
6.     // 下面是配置输入路径  
7.      FileInputFormat.addInputPath(job, new Path(crawlDb, CrawlDb.CURRENT_NAME));  
8.      job.setInputFormat(SequenceFileInputFormat.class);   // 这里配置输入的文件格式，这里实际上是MapSequenceFileInputFormat,这不过这两个是通用的  
9.     
10.     
11.      job.setMapperClass(CrawlDbStatMapper.class);         // 这里配置Mapper方法   
12.    job.setCombinerClass(CrawlDbStatCombiner.class);       // 这里配置一个Combiner方法，这个方法主要是在Mapper端对数据进行聚合，起到了优化作用  
13.    job.setReducerClass(CrawlDbStatReducer.class);         // 这里配置了相应的Reduce方法  
14.     
15.     
16.    FileOutputFormat.setOutputPath(job, tmpFolder);        // 输出目录  
17.    job.setOutputFormat(SequenceFileOutputFormat.class);   // 输出的文件格式  
18. // 下面是输出的<key,value>的类型，这里为<Text,LongWritable> ，  
19. // NOTE:这里补充说明一下，所有通过OutputFormat的数据都要继承自Hadoop的序列化框架接口Writable，用于把这个对象  
20. // 序列化与反序列化，这里的Text这是继承自Writable接口的，所以你自定义的抽象类型要使用Hadoop的架构写出到文件中的话，  
21. // 一定要记得继承Writable接口  
22.    job.setOutputKeyClass(Text.class);  
23.    job.setOutputValueClass(LongWritable.class);  
24.     
25.     
26.    JobClient.runJob(job);  // 提交任务  

下面来看一下CrawlDbStatMapper方法做了些什么，部分源代码如下：
 

 

1. public void map(Text key, CrawlDatum value, OutputCollector<Text, LongWritable> output, Reporter reporter)  
2.             throws IOException {  
3.       output.collect(new Text("T"), COUNT_1);   // 这里统计所有url出现的次数，在Reduce端会对这个key T进行聚合，不过在  
4.       // 每一个Map端会调用相应的Combiner进行本地聚合来优化  
5.       output.collect(new Text("status " + value.getStatus()), COUNT_1);  // 来统计每一种url状态的个数  
6.       output.collect(new Text("retry " + value.getRetriesSinceFetch()), COUNT_1); // 这里统计被重新抓取的url个数  
7.       output.collect(new Text("s"), new LongWritable((long) (value.getScore() * 1000.0))); // 这个统计所有url的分数这和  
8.       if(sort){ // 这个参数是configure方法得到的，代码：sort = job.getBoolean("db.reader.stats.sort", false );  
9.         URL u = new URL(key.toString());  
10.         String host = u.getHost();  
11.         output.collect(new Text("status " + value.getStatus() + " " + host), COUNT_1);  // 这里统计相同状态,相同host的url的个数  
12.       }  
13.     }  

    这里来看一下CrawlDbStatCombiner做了些什么，这个类是继承自Reducer这个抽象类的，部分代码如下：
  

 

1. if (!k.equals("s")) {  // 这里统计除分数外的所有相同key的value值之各  
2.     while (values.hasNext()) {  
3.       LongWritable cnt = (LongWritable)values.next();  
4.       val.set(val.get() + cnt.get());  
5.     }  
6.     output.collect(key, val);  
7.   } else {  // 这里统计相同key的分数之和,这里的key就是上面Mapper中输出的's'，就是所有url的分数  
8.     long total = 0;  
9.     long min = Long.MAX_VALUE;  
10.     long max = Long.MIN_VALUE;  
11.     while (values.hasNext()) {  
12.       LongWritable cnt = (LongWritable)values.next();  
13.       if (cnt.get() < min) min = cnt.get();   // 计算最小分数  
14.       if (cnt.get() > max) max = cnt.get();   // 计算最大分数  
15.       total += cnt.get();                    // 计算总分  
16.     }  
17.     output.collect(new Text("scn"), new LongWritable(min)); // 输出这个Mapper节点上的最小分数  
18.     output.collect(new Text("scx"), new LongWritable(max)); // 输出这个Mapper节点上的最大分数  
19.     output.collect(new Text("sct"), new LongWritable(total)); // 输出这个Mapper节点上的总分  
20.   }  
21. }  

Hadoop中的Combiner主要的作用是对Mapper的输出进行次优化，以减少Reducer的scoket的网络传输数据量

最后看一下那个CrawlDbStatReducer方法，主要代码如下：

 

1. String k = ((Text) key).toString();  
2.      if (k.equals("T")) {            // 这里统计所有url的个数  
3.        // sum all values for this key  
4.        long sum = 0;  
5.        while (values.hasNext()) {  
6.          sum += ((LongWritable) values.next()).get();  
7.        }  
8.        // output sum  
9.        output.collect(key, new LongWritable(sum));  
10.      } else if (k.startsWith("status") || k.startsWith("retry")) {// 这里统计所有key中包含"status"与"retry"字段的value的值  
11.        LongWritable cnt = new LongWritable();  
12.        while (values.hasNext()) {  
13.          LongWritable val = (LongWritable)values.next();  
14.          cnt.set(cnt.get() + val.get());  
15.        }  
16.        output.collect(key, cnt);  
17.      } else if (k.equals("scx")) {               // 这里计算url分数的max值  
18.        LongWritable cnt = new LongWritable(Long.MIN_VALUE);  
19.        while (values.hasNext()) {  
20.          LongWritable val = (LongWritable)values.next();  
21.          if (cnt.get() < val.get()) cnt.set(val.get());  
22.        }  
23.        output.collect(key, cnt);  
24.      } else if (k.equals("scn")) {              // 这里统计url分数的min值  
25.        LongWritable cnt = new LongWritable(Long.MAX_VALUE);  
26.        while (values.hasNext()) {  
27.          LongWritable val = (LongWritable)values.next();  
28.          if (cnt.get() > val.get()) cnt.set(val.get());  
29.        }  
30.        output.collect(key, cnt);  
31.      } else if (k.equals("sct")) {             // 这里统计所有url的总分  
32.        LongWritable cnt = new LongWritable();  
33.        while (values.hasNext()) {  
34.          LongWritable val = (LongWritable)values.next();  
35.          cnt.set(cnt.get() + val.get());  
36.        }  
37.        output.collect(key, cnt);  
38.      }  

在processStatJob这个方法中最所还要读取上面MP的输出，对其数据进行规格化输出，这里就不分析了，只是要注意一点，代码如下

 

1. // reading the result  
2.    FileSystem fileSystem = FileSystem.get(config);  
3.    SequenceFile.Reader[] readers = SequenceFileOutputFormat.getReaders(config, tmpFolder);  

        这里使用了SequenceFileOutputFormat的Reader来读取其内容，在Reader中有一个叫next(key,value)的方法来读取相应的<key,value>对

    * bin/nutch <crawldb> -dump <out_dir> [-format normal|csv]

      这个命令是调用CrawlDbReader中的processDumpJob方法来做的，这个方法也是提交了一个MP任务，不过这个MP任务相对来说很简单了，就是定义了InputFormat与OutputFormat，没有定义Map与Reducer操作，来对CrawlDb的数据进行转换。

    * bin/nutch <crawldb> -url url
 这个命令是调用CrawlDbReader中的get方法来做的，主要用到了两个方法，一个是：

 

1. private void openReaders(String crawlDb, Configuration config) throws IOException {  
2.     if (readers != null) return;  
3.     FileSystem fs = FileSystem.get(config);  
4.     readers = MapFileOutputFormat.getReaders(fs, new Path(crawlDb,  
5.         CrawlDb.CURRENT_NAME), config);  
6.     }  

 这个方法主要是用于得到相应的Readers抽象，输入为CrawlDb的目录名，这个目录的文件格式为MapFileOutputFormat

 另一个方法是：

 

1. CrawlDatum res = (CrawlDatum)MapFileOutputFormat.getEntry(readers,  
2.       new HashPartitioner<Text, CrawlDatum>(), key, val);  

 根据上面得到的readers，通过MapFileOutputFormat的static方法getEntry来得到key(url)所对应的value(CrawlDatum),注意这里getEntry返回的类型为Writable抽象，这里要cast一下. 

* bin/nutch <crawldb> -topN nnn <out_dir> [<min>]

 这个命令是调用CrawlDbReader中的processTopNJob方法来估物，这里也使用两个MP任务来做。
 第一个MP任务主要是把<url,CrawlDatum>转换成<score,url>的格式，过滤那些score小于min的数据，用来准备数据
 第二个MP任务主要是用来对score进行排序，产生topN个url，输出到一个文件中，因为这里设置了job.setNumReduceTasks(1).

 NOTE:这里为什么要用两个MP任务，主要是因为这里的key是score，如果用一个MP任务，那在Reducer的时候就会对相同的key进行聚合，这时就会出问题，而这里第一个MP任务是不对key进行聚合的，这里使用了IdnetityReducer。


3. 总结
   这个是工具类，主要做一些统计工作

作者：http://blog.csdn.net/amuseme_lu

---

相关文章阅读及免费下载：

《Apache Nutch 1.3 学习笔记目录》

《Apache Nutch 1.3 学习笔记一》

《Apache Nutch 1.3 学习笔记二》

《Apache Nutch 1.3 学习笔记三（Inject）》

《Apache Nutch 1.3 学习笔记三（Inject CrawlDB Reader）》

《Apache Nutch 1.3 学习笔记四（Generate）》

《Apache Nutch 1.3 学习笔记四（SegmentReader分析）》

《Apache Nutch 1.3 学习笔记五（FetchThread）》

《Apache Nutch 1.3 学习笔记五（Fetcher流程）》

《Apache Nutch 1.3 学习笔记六（ParseSegment）》

《Apache Nutch 1.3 学习笔记七（CrawlDb - updatedb）》

《Apache Nutch 1.3 学习笔记八（LinkDb）》

《Apache Nutch 1.3 学习笔记九（SolrIndexer）》

《Apache Nutch 1.3 学习笔记十（Ntuch 插件机制简单介绍）》

《Apache Nutch 1.3 学习笔记十（插件扩展）》

《Apache Nutch 1.3 学习笔记十（插件机制分析）》

《Apache Nutch 1.3 学习笔记十一（页面评分机制 OPIC）》

《Apache Nutch 1.3 学习笔记十一（页面评分机制 LinkRank 介绍）》

《Apache Nutch 1.3 学习笔记十二（Nutch 2.0 的主要变化）》

更多《Apache Nutch文档》，尽在开卷有益360 http://www.docin.com/book_360