Nutch源码阅读进程2---Generate

继之前仓促走完nutch的第一个流程Inject后，再次起航，Debug模式走起，进入第二个预热阶段Generate~~~

上期回顾：Inject主要是将爬取列表中的url转换为指定格式<Text，CrawlDatum>存在CrawlDb中，主要做了两件事，一是读取种子列表中的url，对其进行了url过滤、规范化，当然这其中用的是hadoop的mapreduce模式提交job到jobtracker，因为没有研读hadoop源码，所以这块先放放，理清nutch的大体思路后再去啃hadoop的mapreduce，总之这是第一个点，随后是将第一个任务执行完的输出作为输入执行第二个任务，主要是判定当前的CrawlDb中的url和要更新的url是否有重复的，通过相应的判断和标记状态（如STATUS_INJECTED、STATUS_DB_UNFETCHED）确保crawldb中此次的Inject的url不会重复，为下一步Generate做准备。

1.首先根据用户输入的depth进行循环，然后伴随了赋了一些必要的值就直接奔着generator.generate(crawlDb, segments, -1, topN, System.currentTimeMillis());方法去了，进入该方法后，首先就是将存放临时文件的目录结构生成，然后生成文件锁Path lock = new Path(dbDir, CrawlDb.LOCK_NAME);

小插曲：其中涉及到对于获取当前时间并转换为我们熟知的格式的代码：

SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
long start = System.currentTimeMillis();

SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
long start = System.currentTimeMillis();
LOG.info("Generator: starting at " + sdf.format(start));

后面就是初始化文件系统、job等，在进行相应的赋值比如输入mapper、reducer、partition等。

注意：这里的输入时前面一个环节Inject产生的Crawldb，输出存放在刚刚生成的tempdir的临时文件夹；

2.随后进入任务的提交阶段，即执行：JobClient.runJob(job);，进入该方法后，走的还是Inject提交job的那一套，包括初始化JobClient，判定是否是local模式以及确定map的个数等，这是第一个走hadoop的任务，其中的mapper、partition和reducer都是Selector类：

job.setMapperClass(Selector.class);
job.setPartitionerClass(Selector.class);
job.setReducerClass(Selector.class);（备注：

在Hadoop Map/Reduce框架下，当Mapper处理好数据后，需要使用Partitioner确定怎样合理地将Mapper输出分配到Reducer之中。

默认的情况下，Hadoop对<key,value>键值对中的key取hash值来确定怎样分配给相应的Reducer。Hadoop使用HashParitioner class来执行这一操作。但是有时候HashPartitioner并不能完成它的功能。）

（插曲：通过调试发现过程中，提交job执行后再runningjob方法中返回的rj值为：

Job: job_local_0003
file: file:/tmp/hadoop-zjhadoop/mapred/staging/zjhadoop2112303622/.staging/job_local_0003/job.xml
tracking URL: http://localhost:8080/
map() completion: 1.0
reduce() completion: 1.0）

代码中的mapper主要完成以下几件工作：（1）判断是否有filter设置，如果有的话则对url进行过滤；（2）通过读取Inject生成的数据中的CrawlDatum字段中的时间加上超时时间决定是否抓取；

if (oldGenTime.get() + genDelay > curTime) // still wait for
// update
return;

（3）计算url的分值，对小于阈值的过滤掉；

sort = scfilters.generatorSortValue((Text) key, crawlDatum, sort);//计算分值

if (scoreThreshold != Float.NaN && sort < scoreThreshold) return;//判定计算出的分值与阈值决定是否要过滤掉

（4）收集未被过滤的url，处理输出为<FloatWritable,SelectorEntry>格式

entry.datum = crawlDatum;
entry.url = (Text) key;//crawlDatum和key共同封装好就是entry了，
output.collect(sortValue, entry); // invert for sort by score 然后再将entry与计算的sortValue组合就是输出类型<FloatWritable,SelectorEntry>了

这些步骤就组成了整个map的工作了；

3.整个Selector是实现了map、partitioner和reducer的，下面还有partioner的功能实现，

Selector中的Partition方法主要是调用了URLPartition来进行相应的分块操作

这里会首先根据url的hashCode来进行partition,如果用户设置了根据domain或者ip来进行partition，那这里会根据用户的配置来进行相应的partition操作；

后面就是reducer模块了，其主要是将没有被过滤的url进行计算，对于每个reducer如果超过一个限定值limit的话就会分开，放到另个segments中

4.下面就是第二大块，用的也是hadoop的mapreducer。紧接着上面，首先执行：

FileStatus[] status = fs.listStatus(tempDir);//该行代码是获取第一个job提交后生成的tempDir文件夹中的信息即读取tempDir的多个fetchlist的segment

读取经过重重检验选拔出来的url，生成segments

Path subfetchlist = stat.getPath();//读取fetchlist的segment

然后再进入方法：Path newSeg = partitionSegment(fs, segments, subfetchlist, numLists);在该方法中又是通过提交一个job解决。其中i输入是刚刚临时文件夹tempDir中的fetchlist，输出是在代码中定义好的output，即类似于crawl20140727/segments/20140727195735/crawl_generate这样的目录结构。（插曲：该job在runningJob方法中执行后返回的rj值为：

Job: job_local_0004
file: file:/tmp/hadoop-zjhadoop/mapred/staging/zjhadoop1993184312/.staging/job_local_0004/job.xml
tracking URL: http://localhost:8080/
map() completion: 1.0
reduce() completion: 1.0）

网上关于partitionSegment的比较详细的解释：（

// invert again, partition by host/domain/IP, sort by url hash

// 从代码的注释中我们可以看到，这里主要是对url按host/domain/IP进行分类

// NOTE：这里的分类就是Partition的意思，就是相同host或者是domain或者是IP的url发到同一台机器上

// 这里主要是通过URLPartitioner来做的，具体是按哪一个来分类，是通用参数来配置的，这里有PARTITION_MODE_DOMAIN，PARTITION_MODE_IP

// 来配置，默认是按Url的hashCode来分。

if (LOG.isInfoEnabled()) {

LOG.info("Generator: Partitioning selected urls for politeness.");

}

Path segment = new Path(segmentsDir, generateSegmentName()); // 也是在segmentDir目录产生一个新的目录，以当前时间命名

Path output = new Path(segment, CrawlDatum.GENERATE_DIR_NAME); // 在上面的目录下再生成一个特定的crawl_generate目录

LOG.info("Generator: segment: " + segment);

/ 下面又用一个MP任务来做

NutchJob job = new NutchJob(getConf());

job.setJobName("generate: partition " + segment);

job.setInt("partition.url.seed", new Random().nextInt()); // 这里产生一个Partition的随机数

FileInputFormat.addInputPath(job, inputDir); // 输入目录名

job.setInputFormat(SequenceFileInputFormat.class); // 输入文件格式

job.setMapperClass(SelectorInverseMapper.class); // 输入的Mapper，主要是过滤原来的key,使用url来做为新的key值

job.setMapOutputKeyClass(Text.class); // Mapper的key输出类型，这里就是url的类型

job.setMapOutputValueClass(SelectorEntry.class); // Mapper的value的输出类型，这里还是原因的SelectorEntry类型

job.setPartitionerClass(URLPartitioner.class); // 这里的key(url)的Partition使用这个类来做,这个类前面有说明

job.setReducerClass(PartitionReducer.class); // 这里的Reducer类，

job.setNumReduceTasks(numLists); // 这里配置工作的Reducer的个数，也就是生成几个相应的输出文件

FileOutputFormat.setOutputPath(job, output); // 配置输出路径

job.setOutputFormat(SequenceFileOutputFormat.class); // 配置输出格式

job.setOutputKeyClass(Text.class); // 配置输出的key与value的类型

job.setOutputValueClass(CrawlDatum.class); // 注意这里返回的类型为<Text,CrawlDatum>

job.setOutputKeyComparatorClass(HashComparator.class); // 这里定义控制key排序的比较方法

JobClient.runJob(job); // 提交任务

return segment; ）

执行完成这个job后也就得到了输出，即一个segments目录，类似于：crawl20140727/segments/20140727195735。后面就是一些清理现场的工作，比如解除文件锁、删除之前创建的临时文件夹等。（这是个好习惯，以后学着点，吃完东西要把嘴擦干净^_^）

5.执行完上步后，就生成了相应的segments文件目录，下一步也是一个mapreduce过程，so，没研究过mapreducer的真是伤不起啊……这个过程主要是更新crawldb数据，保证下次generate不会有相同的url。其中mapreduce都是crawlDbUpdate类：

job.setMapperClass(CrawlDbUpdater.class);
job.setReducerClass(CrawlDbUpdater.class);

至此Nutch的第二步已经走完，完事具备，只欠fetch~~~~

虽然源码看着有些头疼，但是很是坚持走下来吧，先整体把我，再细细研读吧，come on！！！

自身能力有限，见解之处必有不足，还望见谅。

参考博文：http://blog.csdn.net/amuseme_lu/article/details/6720079

友情赞助

如果你觉得博主的文章对你那么一点小帮助，恰巧你又有想打赏博主的小冲动，那么事不宜迟，赶紧扫一扫，小额地赞助下，攒个奶粉钱，也是让博主有动力继续努力，写出更好的文章^^。

　　　　1. 支付宝　　　　　　　　　　　　　　　　　　　　　　　　　　2. 微信

Nutch源码阅读进程2---Generate 　　　　　　　　　　　　　　　　　　　　　　

秒客网

Nutch源码阅读进程2---Generate

Selector中的Partition方法主要是调用了URLPartition来进行相应的分块操作

相关文章