Inject - Nutch 20 Mar 2014

$ bin/nutch inject <url_dir> <crawl_db>

bin/nutch readdb <crawl_db> -stats -sort

JobConf sortJob = new NutchJob(getConf());                 // 生成一个Nutch的配置抽象
sortJob.setJobName("inject " + urlDir);
FileInputFormat.addInputPath(sortJob, urlDir);             // 设置InputFormat，这里为FileInputFormat,这里要注意的是可以调用多次addInputPath这个方法，效果是会有多个输入源
sortJob.setMapperClass(InjectMapper.class);                // 这里设置了Mapper方法，主要是用于解析、过滤和规格化url文本，把其转换成<url,CrawlDatum>格式
FileOutputFormat.setOutputPath(sortJob, tempDir);          // 这里定义了一个输出路径，这里的tempDir=mapred.temp.dir/inject-temp-Random()
sortJob.setOutputFormat(SequenceFileOutputFormat.class);   // 这里配置了输出格式，这里为SequenceFileOutputFormat，这是MP的一种二进制输出结构
sortJob.setOutputKeyClass(Text.class);                     // 这里配置了MP的输出<key,value>的类型，这里为<Text,CrawlDatum>
sortJob.setOutputValueClass(CrawlDatum.class);
sortJob.setLong("injector.current.time", System.currentTimeMillis());
JobClient.runJob(sortJob);                                 // 这里用于提交任务到JobTracker，让其运行任务

public void map(WritableComparable key, Text value, OutputCollector<Text, CrawlDatum> output, Reporter reporter) throws IOException {
    String url = value.toString();                                            // value is line of text
    if (url != null && url.trim().startsWith("#")) {                          // 这里以#号开头的文本就过滤
        /* Ignore line that start with # */
        return;
    }
    // if tabs : metadata that could be stored
    // must be name=value and separated by \t
    float customScore = -1f;
    int customInterval = interval;
    Map<String,String> metadata = new TreeMap<String,String>();                // 设置属性的一个容器
    if (url.indexOf("\t") != -1) {
        String[] splits = url.split("\t");                                     // 对一行文本进行切分
        url = splits[0];
        for (int s = 1; s < splits.length; s++) {
            // find separation between name and value
            int indexEquals = splits[s].indexOf("=");
            if (indexEquals == -1) {
                // skip anything without a =
                continue;
            }
            String metaname = splits[s].substring(0, indexEquals);              // 得到元数据的名字
            String metavalue = splits[s].substring(indexEquals+1);              // 得到元数据的值
            if (metaname.equals(nutchScoreMDName)) {                            // 看是不是保留的元数据
                try {
                    customScore = Float.parseFloat(metavalue);
                } catch (NumberFormatException nfe) {}
            } else if (metaname.equals(nutchFetchIntervalMDName)) {
                try {
                    customInterval = Integer.parseInt(metavalue);
                } catch (NumberFormatException nfe){}
            } else metadata.put(metaname,metavalue);                             // 如果这个元数据不是保留的元数据，就放到容器中
        }
    }
    try {
        url = urlNormalizers.normalize(url, URLNormalizers.SCOPE_INJECT);        // 对url进行规格化，这里调用的是plugins中的插件
        url = filters.filter(url);                                               // 以url进行过滤
    } catch (Exception e) {
        if (LOG.isWarnEnabled()) { LOG.warn("Skipping " +url + ":" + e); }
        url = null;
    }
    if (url != null) {                                                           // if it passes
        value.set(url);                                                          // collect it // 这里生成一个CrawlDatum对象，设置一些url的初始化数据
        CrawlDatum datum = new CrawlDatum(CrawlDatum.STATUS_INJECTED, customInterval);
        datum.setFetchTime(curTime);                                             // 设置当前url的抓取时间
        // now add the metadata
        Iterator<String> keysIter = metadata.keySet().iterator();
        while (keysIter.hasNext()){                                              // 配置其元数据
            String keymd = keysIter.next();
            String valuemd = metadata.get(keymd);
            datum.getMetaData().put(new Text(keymd), new Text(valuemd));
        } // 设置初始化分数
        if (customScore != -1) datum.setScore(customScore);
        else datum.setScore(scoreInjected);
        try { // 这里对url的分数进行初始化
            scfilters.injectedScore(value, datum);
        } catch (ScoringFilterException e) {
            if (LOG.isWarnEnabled()) {
                LOG.warn("Cannot filter injected score for url " + url
                         + ", using default (" + e.getMessage() + ")");
            }
        }
        // Map 收集相应的数据，类型为<Text,CrawlDatum>
        output.collect(value, datum);
    }
}

// merge with existing crawl db
JobConf mergeJob = CrawlDb.createJob(getConf(), crawlDb);                      // 这里对Job进行相应的配置
FileInputFormat.addInputPath(mergeJob, tempDir);                               // 这里配置了输入的文本数据，就是上面第一个MP任务的输出
mergeJob.setReducerClass(InjectReducer.class);                                 // 这里配置了Reduce的抽象类，这里会覆盖上面createJob设置的Reduce类
JobClient.runJob(mergeJob);                                                    // 提交运行任务
CrawlDb.install(mergeJob, crawlDb);                                            // 把上面新生成的目录重命名为crawlDb的标准文件夹名，然后再删除老的目录
// clean up
FileSystem fs = FileSystem.get(getConf());
fs.delete(tempDir, true);                                                      // 把第一个MP任务的输出目录删除

public static JobConf createJob(Configuration config, Path crawlDb) throws IOException {            // 生成新的CrawlDb文件名
    Path newCrawlDb = new Path(crawlDb,Integer.toString(new Random().nextInt(Integer.MAX_VALUE)));
    JobConf job = new NutchJob(config);                                                             // 生成相应的Job配置抽象
    job.setJobName("crawldb " + crawlDb);
    Path current = new Path(crawlDb, CURRENT_NAME);
    if (FileSystem.get(job).exists(current)) {             // 如果存在老的CrawlDb目录，将其加入InputPath路径中，和上面的tempDir一起进行合并
        FileInputFormat.addInputPath(job, current);
    }
    // NOTE：有没有注意到这里如果有老的CrawlDb目录的话，那它的文件格式是MapFileOutputFormat，
    // 而下面对其读取用了SequenceFileInputFormat来读，因为这两个类底层都是调用了SequenceFile的Reader与Writer来读写的，所以可以通用。
    job.setInputFormat(SequenceFileInputFormat.class);     // 设置CrawlDb目录文件的格式为SequenceFileInputFormat
    job.setMapperClass(CrawlDbFilter.class);               // 设置相应的Map操作，主要是过滤和规格化url
    job.setReducerClass(CrawlDbReducer.class);             // 设置相应的Reduce操作，主要是对url进行聚合
    FileOutputFormat.setOutputPath(job, newCrawlDb);       // 设置新的输出路径
    job.setOutputFormat(MapFileOutputFormat.class);        // 设置输出的格式，这里是MapFileOutputFormat
    job.setOutputKeyClass(Text.class);                     // 这里设置了输出的类型<Text,CrawlDatum>
    job.setOutputValueClass(CrawlDatum.class);
    return job;
}

public void reduce(Text key, Iterator<CrawlDatum> values, OutputCollector<Text, CrawlDatum> output, Reporter reporter) throws IOException {
	// 把相同url聚合后的结果进行处理，这个循环主要是新注入的url与老的url有没有相同的， 
	// 如果有相同的话就不设置其状态，支持collect出去了
    boolean oldSet = false; 
    while (values.hasNext()) {
        CrawlDatum val = values.next();
        if (val.getStatus() == CrawlDatum.STATUS_INJECTED) {
            injected.set(val);
            injected.setStatus(CrawlDatum.STATUS_DB_UNFETCHED);
        } else {
            old.set(val);
            oldSet = true;
        }
    }
    CrawlDatum res = null;
    if (oldSet) res = old; // don't overwrite existing value
    else res = injected;
    output.collect(key, res);
}

public static void install(JobConf job, Path crawlDb) throws IOException {
    Path newCrawlDb = FileOutputFormat.getOutputPath(job);                                         // 得到第二个MP任务的输出目录
    FileSystem fs = new JobClient(job).getFs();
    Path old = new Path(crawlDb, "old");
    Path current = new Path(crawlDb, CURRENT_NAME);                                                // 得到CrawlDb的正规目录名，也就是有没有老的CrawlDb
    if (fs.exists(current)) {                                                                      // 如果有老的CrawlDb目录，把老的目录名生命名为old这个名字
        if (fs.exists(old)) fs.delete(old, true);                                                  // 这里判断old这个目录是不是已经存在，如果存在就删除之
        fs.rename(current, old);
    }
    fs.mkdirs(crawlDb);
    fs.rename(newCrawlDb, current);                                                                // 这里是把第二个MP任务的输出目录重命名为current目录，也就是正规目录名
    if (fs.exists(old)) fs.delete(old, true);                                                      // 删除重使名后的老的CrawlDb目录
    Path lock = new Path(crawlDb, LOCK_NAME);
    LockUtil.removeLockFile(fs, lock);                                                             // 目录解锁
}