Nutch index源代码解析(一)

Nutch集成slor的索引方法介绍

/**

   * 建立索引

   * @param solrUrl solr的web地址

   * @param crawlDb 爬取DB的存放路径：\crawl\crawldb

   * @param linkDb 爬取link的存放路径：\crawl\linkdb

   * @param segments 元数据存放路径：\crawl\segments

   * @param noCommit  是否提交slor服务器跟下slor索引

   * @param deleteGone 是否删除过时的文档

   * @param solrParams solr的参数

   * @param filter 是否启用URL过滤

   * @param normalize 是否格式化 URL

   * @throws IOException

   */

  public void indexSolr(String solrUrl, Path crawlDb, Path linkDb,

      List<Path> segments, boolean noCommit, boolean deleteGone, String solrParams,

      boolean filter, boolean normalize) throws IOException {

        ...

       IndexerMapReduce.initMRJob(crawlDb, linkDb, segments, job);

        ...

   }

Nutch的索引，是通过一个MR完成的。

map的输入是Nutch爬去目录下的sequenceFile ，key是Nutch爬取下来的URL，value使用的java泛型，将所有nutch自定义的数据类型抽象成一个NutchWritable对象。

Nutchwritable的包含的的数据类型如下源代码：

CLASSES = new Class[] {

      org.apache.hadoop.io.NullWritable.class,

      org.apache.hadoop.io.BooleanWritable.class,

      org.apache.hadoop.io.LongWritable.class,

      org.apache.hadoop.io.BytesWritable.class,

      org.apache.hadoop.io.FloatWritable.class,

      org.apache.hadoop.io.IntWritable.class,

      org.apache.hadoop.io.MapWritable.class,

      org.apache.hadoop.io.Text.class,

      org.apache.hadoop.io.MD5Hash.class,

      org.apache.nutch.crawl.CrawlDatum.class,

      org.apache.nutch.crawl.Inlink.class,

      org.apache.nutch.crawl.Inlinks.class,

      org.apache.nutch.fetcher.FetcherOutput.class,

      org.apache.nutch.metadata.Metadata.class,

      org.apache.nutch.parse.Outlink.class,

      org.apache.nutch.parse.ParseText.class,

      org.apache.nutch.parse.ParseData.class,

      org.apache.nutch.parse.ParseImpl.class,

      org.apache.nutch.parse.ParseStatus.class,

      org.apache.nutch.protocol.Content.class,

      org.apache.nutch.protocol.ProtocolStatus.class,

      org.apache.nutch.scoring.webgraph.LinkDatum.class,

    };

这些数据类型分别抽象了Nutch在爬取时各个阶段的数据类型。

map阶段不对value进行处理，只对URL进行处理，处理代码如下：

   String urlString = filterUrl(normalizeUrl(key.toString()));

调用是对URL根据定义好的过滤规则对URL进行过滤和格式化，当然是否进行这步操作可以通过调用命令时参数设置。

reduce是对所有爬取的数据进行处理，代码注释如下：

 /**

   * 输出格式：url作为key，索引的action作为value

   */

  public void reduce(Text key, Iterator<NutchWritable> values,

                     OutputCollector<Text, NutchIndexAction> output, Reporter reporter)

    throws IOException {

    Inlinks inlinks = null;

    CrawlDatum dbDatum = null;

    CrawlDatum fetchDatum = null;

    ParseData parseData = null;

    ParseText parseText = null;

 

    while (values.hasNext()) {

      final Writable value = values.next().get(); // unwrap

      //如果是URL注入的数据类型

      if (value instanceof Inlinks) {

        inlinks = (Inlinks)value;

        //如果是爬取的数据类型

      } else if (value instanceof CrawlDatum) {

        final CrawlDatum datum = (CrawlDatum)value;

        //如果当前数据处于db注入状态

        if (CrawlDatum.hasDbStatus(datum)) {

          dbDatum = datum;

        }

        //如果当前数据处于爬取完成状态。

        else if (CrawlDatum.hasFetchStatus(datum)) {

 

          // don't index unmodified (empty) pages

          //判断爬去的是否进行了修改

          if (datum.getStatus() != CrawlDatum.STATUS_FETCH_NOTMODIFIED) {

            fetchDatum = datum;

 

            /**

             * Check if we need to delete 404 NOT FOUND and 301 PERMANENT REDIRECT.

             */

            //参数中如果设置删除为true则删除错误及过时的页面

            if (delete) {

            //如果爬取的页面过期，采取删除操作。

              if (fetchDatum.getStatus() == CrawlDatum.STATUS_FETCH_GONE) {

                reporter.incrCounter("IndexerStatus", "Documents deleted", 1);

 

                NutchIndexAction action = new NutchIndexAction(null, NutchIndexAction.DELETE);

                output.collect(key, action);

                return;

              }

              //如果爬去的页面已经重定向到另外一个页面，才去删除操作。

              if (fetchDatum.getStatus() == CrawlDatum.STATUS_FETCH_REDIR_PERM) {

                reporter.incrCounter("IndexerStatus", "Perm redirects deleted", 1);

 

                NutchIndexAction action = new NutchIndexAction(null, NutchIndexAction.DELETE);

                output.collect(key, action);

                return;

              }

            }

          }

         //URL是通过其他URL被发现的 ||页面的签名||页面的元数据是通过解析器产生的

        } else if (CrawlDatum.STATUS_LINKED == datum.getStatus() ||

                   CrawlDatum.STATUS_SIGNATURE == datum.getStatus() ||

                   CrawlDatum.STATUS_PARSE_META == datum.getStatus()) {

          continue;

        } else {

          throw new RuntimeException("Unexpected status: "+datum.getStatus());

        }

       //如果是解析的数据类型

      } else if (value instanceof ParseData) {

        parseData = (ParseData)value;

 

        // Handle robots meta? https://issues.apache.org/jira/browse/NUTCH-1434

        if (deleteRobotsNoIndex) {

          // Get the robots meta data

          String robotsMeta = parseData.getMeta("robots");

 

          // Has it a noindex for this url?

          if (robotsMeta != null && robotsMeta.toLowerCase().indexOf("noindex") != -1) {

            // Delete it!

            NutchIndexAction action = new NutchIndexAction(null, NutchIndexAction.DELETE);

            output.collect(key, action);

            return;

          }

        }

        //解析完的Text文件

      } else if (value instanceof ParseText) {

        parseText = (ParseText)value;

      } else if (LOG.isWarnEnabled()) {

        LOG.warn("Unrecognized type: "+value.getClass());

      }

    }

    //如果只有链接，没有爬取历史的记录或者爬取数据直接返回

    if (fetchDatum == null || dbDatum == null

        || parseText == null || parseData == null) {

      return;                                     // only have inlinks

    }

 

    // Whether to skip DB_NOTMODIFIED pages

    //如果页面被爬取过，但是没有进行修过，在传进来的命令中设置了跳过则跳过。

    if (skip && dbDatum.getStatus() == CrawlDatum.STATUS_DB_NOTMODIFIED) {

      reporter.incrCounter("IndexerStatus", "Skipped", 1);

      return;

    }

    //页面爬去成功，但是解析失败的，直接返回

    if (!parseData.getStatus().isSuccess() ||

        fetchDatum.getStatus() != CrawlDatum.STATUS_FETCH_SUCCESS) {

      return;

    }

 

    NutchDocument doc = new NutchDocument();

    //解析完的数据中获取页面的元数据

    final Metadata metadata = parseData.getContentMeta();

 

    // add segment, used to map from merged index back to segment files

    doc.add("segment", metadata.get(Nutch.SEGMENT_NAME_KEY));

    //页面摘要

    // add digest, used by dedup

    doc.add("digest", metadata.get(Nutch.SIGNATURE_KEY));

    

    final Parse parse = new ParseImpl(parseText, parseData);

    try {

      // extract information from dbDatum and pass it to

      // fetchDatum so that indexing filters can use it

      final Text url = (Text) dbDatum.getMetaData().get(Nutch.WRITABLE_REPR_URL_KEY);

      if (url != null) {

        fetchDatum.getMetaData().put(Nutch.WRITABLE_REPR_URL_KEY, url);

      }

      // run indexing filters

      //执行所有过滤器

      doc = this.filters.filter(doc, parse, key, fetchDatum, inlinks);

    } catch (final IndexingException e) {

      if (LOG.isWarnEnabled()) { LOG.warn("Error indexing "+key+": "+e); }

      reporter.incrCounter("IndexerStatus", "Errors", 1);

      return;

    }

 

    // skip documents discarded by indexing filters

    if (doc == null) {

      reporter.incrCounter("IndexerStatus", "Skipped by filters", 1);

      return;

    }

 

    float boost = 1.0f;

    // run scoring filters

    //执行评分过滤器

    try {

      boost = this.scfilters.indexerScore(key, doc, dbDatum,

              fetchDatum, parse, inlinks, boost);

    } catch (final ScoringFilterException e) {

      if (LOG.isWarnEnabled()) {

        LOG.warn("Error calculating score " + key + ": " + e);

      }

      return;

    }

    //将评分作为文档的权重

    // apply boost to all indexed fields.

    doc.setWeight(boost);

    // store boost for use by explain and dedup

    doc.add("boost", Float.toString(boost));

 

    reporter.incrCounter("IndexerStatus", "Documents added", 1);

 

    NutchIndexAction action = new NutchIndexAction(doc, NutchIndexAction.ADD);

    output.collect(key, action);

  }

目前研究到此，后面的待续。。