一、主程序調(diào)用

SolrDeleteDuplicates dedup = new SolrDeleteDuplicates();

dedup.setConf(getConf());

dedup.dedup(solrUrl);

二、job任務(wù)配置

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任務(wù)的輸入和輸出

Map任務(wù)輸入、輸出

public void map(

K key, V val,

      OutputCollector<K, V> output

reduce任務(wù)輸入、輸出

輸入：Text/Iterator<SolrRecord>

輸出：Text/SolrRecord

public void reduce(

Text key, Iterator<SolrRecord> values,

      OutputCollector<Text, SolrRecord> output

四、job任務(wù)輸入類SolrInputFormat

getSplits方法將所有的文檔按照數(shù)量平均分片

getRecordReader方法中利用solrserver查詢了當(dāng)前分片包含的所有doc記錄，solrrecord返回了的當(dāng)前的RecordReader<Text, SolrRecord>記錄（RecordReader是一個全局的變量），并且有獲取下一個方法。

(1)、SolrInputFormat的getSplits方法

1、根據(jù)job對象的參數(shù)，獲取solrserver對象。

2、構(gòu)建并執(zhí)行查詢（查詢參數(shù)：[*:*、id、setRow(1)] ），獲取響應(yīng)對象

3、根據(jù)響應(yīng)對象獲取索引總數(shù)，除以分片數(shù)，得到每一片分配多少個索引

4、根據(jù)分片數(shù)創(chuàng)建 SolrInputSplit數(shù)組對象，

5、根據(jù)solr輸入分片的開始和結(jié)束位置，實例化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參數(shù)，強轉(zhuǎn)成SolrInputSplit對象，并獲取這個分片的文檔總數(shù)

3、構(gòu)建查詢對象，執(zhí)行查詢（參數(shù)[*:*，id，boost，tstamp，digest， SolrInputSplit中的開始位置，文檔總數(shù) ]）。

4、根據(jù)響應(yīng)對象，獲取結(jié)果集

5、對匿名內(nèi)部內(nèi)RecordReader做了實現(xiàn)，并且返回

    public RecordReader<Text, SolrRecord> getRecordReader(final InputSplit split,

        final JobConf job, 

        Reporter reporter)

        throws IOException {

//1、獲取solrserver對象

      SolrServer solr = SolrUtils.getCommonsHttpSolrServer(job);

//2、將傳入的split參數(shù)，強轉(zhuǎn)成SolrInputSplit對象，并獲取這個分片的文檔總數(shù)

SolrInputSplit solrSplit = (SolrInputSplit) split;

      final int numDocs = solrSplit.getNumDocs();

//3、構(gòu)建查詢對象，執(zhí)行查詢（參數(shù)[*:*，id，boost，tstamp，digest，

SolrInputSplit中的開始位置，文檔總數(shù)

]）

      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、根據(jù)響應(yīng)對象，獲取結(jié)果集

      final SolrDocumentList solrDocs = response.getResults();

      return new RecordReader<Text, SolrRecord>() {

//當(dāng)前的文檔

        private int currentDoc = 0;

        public void close() throws IOException { }

        public Text createKey() {

          return new Text();

        }

        public SolrRecord createValue() {

          return new SolrRecord();

        }

//獲取當(dāng)前的指針

        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()方法中的實現(xiàn)

(1)、map任務(wù)

(2)、reduce任務(wù)

去重邏輯：

reduce任務(wù)會遍歷每一個record，并執(zhí)行reduce()方法中的代碼

reduce()方法中，會遍歷處于當(dāng)前文檔之后的所有文檔，如果分值和時間都比當(dāng)前的小，會調(diào)用solrj刪除這個文檔，如果比當(dāng)前的大，會刪除當(dāng)前的，并把當(dāng)前的替換成這個大的。

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

      OutputCollector<Text, SolrRecord> output, Reporter reporter)

  throws IOException {

     //1、下一個SolrRecord對象

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

//2、遍歷了SolrRecord

    while (values.hasNext()) {

//

      SolrRecord solrRecord = values.next();

//boost、tstamp參與比較

//如果當(dāng)前的分值， 比保持的分支高，并且時間比保持的新，就根據(jù)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;

      }

    }

  }

六、關(guān)于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();