nutch中如何实现索引去重

一、主程序调用

SolrDeleteDuplicates dedup = new SolrDeleteDuplicates();

dedup.setConf(getConf());

dedup.dedup(solrUrl);

二、job任务配置

JobConf job = new NutchJob(getConf());

job.setInputFormat(SolrInputFormat.class);

job.setMapperClass(IdentityMapper.class);

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(SolrRecord.class);

job.setReducerClass(SolrDeleteDuplicates.class);

job.setOutputFormat(NullOutputFormat.class);

JobClient.runJob(job);

三、Map、reduce任务的输入和输出

Map任务输入、输出

public void map(

K key, V val,

      OutputCollector output

reduce任务输入、输出

输入：Text/Iterator

输出：Text/SolrRecord

public void reduce(

Text key, Iterator values,

      OutputCollector output

四、job任务输入类SolrInputFormat

getSplits方法将所有的文档按照数量平均分片

getRecordReader方法中利用solrserver查询了当前分片包含的所有doc记录，solrrecord返回了的当前的RecordReader记录（RecordReader是一个全局的变量），并且有获取下一个方法。

(1)、SolrInputFormat的getSplits方法

1、根据job对象的参数，获取solrserver对象。

2、构建并执行查询（查询参数：[*:*、id、setRow(1)] ），获取响应对象

3、根据响应对象获取索引总数，除以分片数，得到每一片分配多少个索引

4、根据分片数创建 SolrInputSplit数组对象，

5、根据solr输入分片的开始和结束位置，实例化SolrInputSplit对象

    public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {

      SolrServer solr = SolrUtils.getCommonsHttpSolrServer(job);

      final SolrQuery solrQuery = new SolrQuery(SOLR_GET_ALL_QUERY);

      solrQuery.setFields(SolrConstants.ID_FIELD);

      solrQuery.setRows(1);

      QueryResponse response;

      try {

        response = solr.query(solrQuery);

      } catch (final SolrServerException e) {

        throw new IOException(e);

      }

      int numResults = (int)response.getResults().getNumFound();

      int numDocsPerSplit = (numResults / numSplits); 

      int currentDoc = 0;

      SolrInputSplit[] splits = new SolrInputSplit[numSplits];

      for (int i = 0; i < numSplits - 1; i++) {

        splits[i] = new SolrInputSplit(currentDoc, numDocsPerSplit);

        currentDoc += numDocsPerSplit;

      }

      splits[splits.length - 1] = new SolrInputSplit(currentDoc, numResults - currentDoc);

      return splits;

    }

(2)、SolrInputFormat的getRecordReader()方法

1、获取solrserver对象

2、将传入的split参数，强转成SolrInputSplit对象，并获取这个分片的文档总数

3、构建查询对象，执行查询（参数[*:*，id，boost，tstamp，digest， SolrInputSplit中的开始位置，文档总数 ]）。

4、根据响应对象，获取结果集

5、对匿名内部内RecordReader做了实现，并且返回

    public RecordReader getRecordReader(final InputSplit split,

        final JobConf job, 

        Reporter reporter)

        throws IOException {

//1、获取solrserver对象

      SolrServer solr = SolrUtils.getCommonsHttpSolrServer(job);

//2、将传入的split参数，强转成SolrInputSplit对象，并获取这个分片的文档总数

SolrInputSplit solrSplit = (SolrInputSplit) split;

      final int numDocs = solrSplit.getNumDocs();

//3、构建查询对象，执行查询（参数[*:*，id，boost，tstamp，digest，

SolrInputSplit中的开始位置，文档总数

]）

      SolrQuery solrQuery = new SolrQuery(SOLR_GET_ALL_QUERY);

      solrQuery.setFields(SolrConstants.ID_FIELD, SolrConstants.BOOST_FIELD,

                          SolrConstants.TIMESTAMP_FIELD,

                          SolrConstants.DIGEST_FIELD);

      solrQuery.setStart(solrSplit.getDocBegin());

      solrQuery.setRows(numDocs);

      QueryResponse response;

try {

        response = solr.query(solrQuery);

      } catch (final SolrServerException e) {

        throw new IOException(e);

      }

//4、根据响应对象，获取结果集

      final SolrDocumentList solrDocs = response.getResults();

      return new RecordReader() {

//当前的文档

        private int currentDoc = 0;

        public void close() throws IOException { }

        public Text createKey() {

          return new Text();

        }

        public SolrRecord createValue() {

          return new SolrRecord();

        }

//获取当前的指针

        public long getPos() throws IOException {

          return currentDoc;

        }

//获取进度

        public float getProgress() throws IOException {

          return currentDoc / (float) numDocs;

        }

//获取下一个

        public boolean next(Text key, SolrRecord value) throws IOException {

          if (currentDoc >= numDocs) {

            return false;

          }

//

          SolrDocument doc = solrDocs.get(currentDoc);

//获取摘要

          String digest = (String) doc.getFieldValue(SolrConstants.DIGEST_FIELD);

//把摘要作为key

          key.set(digest);

//value（SolrRecord）

//赋值：通过doc给solrrecord的id，tstamp，boost 3个字段赋值

          value.readSolrDocument(doc);

//指针加自增1

          currentDoc++;

          return true;

        }    

      };

    }

五、map()方法和reduce()方法中的实现

(1)、map任务

(2)、reduce任务

去重逻辑：

reduce任务会遍历每一个record，并执行reduce()方法中的代码

reduce()方法中，会遍历处于当前文档之后的所有文档，如果分值和时间都比当前的小，会调用solrj删除这个文档，如果比当前的大，会删除当前的，并把当前的替换成这个大的。

  public void reduce(Text key, Iterator values,

      OutputCollector output, Reporter reporter)

  throws IOException {

     //1、下一个SolrRecord对象

    SolrRecord recordToKeep = new SolrRecord(values.next());

//2、遍历了SolrRecord

    while (values.hasNext()) {

//

      SolrRecord solrRecord = values.next();

//boost、tstamp参与比较

//如果当前的分值， 比保持的分支高，并且时间比保持的新，就根据id删除这条索引，

      if (solrRecord.getBoost() > recordToKeep.getBoost() ||

          (solrRecord.getBoost() == recordToKeep.getBoost() && 

              solrRecord.getTstamp() > recordToKeep.getTstamp())) {

        updateRequest.deleteById(recordToKeep.id);

        recordToKeep = new SolrRecord(solrRecord);

      } else {

        updateRequest.deleteById(solrRecord.id);

      }

      numDeletes++;

      reporter.incrCounter("SolrDedupStatus", "Deleted documents", 1);

      if (numDeletes >= NUM_MAX_DELETE_REQUEST) {

        try {

          LOG.info("SolrDeleteDuplicates: deleting " + numDeletes + " duplicates");

          updateRequest.process(solr);

        } catch (SolrServerException e) {

          throw new IOException(e);

        }

        updateRequest = new UpdateRequest();

        numDeletes = 0;

      }

    }

  }

六、关于digest

doc中的digest字段，是在IndexerMapReduce类中的reduce方法中加入的

// add digest, used by dedup

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

Metadata中包含了一个HashMap

final Metadata metadata = parseData.getContentMeta();