nutch SolrDeleteDuplicates

map   使用默认的map
  InputFormat 负责split数据转换数据
job.setInputFormat(SolrInputFormat.class);
SolrInputFormat 会根据分配多个split做数据切分
如下
  /** Return each index as a split. */
    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;
    }
getRecordReader 会根据split把自己的数据取出来
next方法遍历数据
    public RecordReader<Text, SolrRecord> getRecordReader(final InputSplit split,
        final JobConf job,
        Reporter reporter)
        throws IOException {

      SolrServer solr = SolrUtils.getCommonsHttpSolrServer(job);
      SolrInputSplit solrSplit = (SolrInputSplit) split;
      final int numDocs = solrSplit.getNumDocs();
     
      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);
      }

      final SolrDocumentList solrDocs = response.getResults();

      return new RecordReader<Text, SolrRecord>() {

        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.set(digest);
          value.readSolrDocument(doc);

          currentDoc++;
          return true;
        }   
      };
    }
  }

reduce 比较相同key的数据，如果有多个数据删除权重小的，或者如果权重相等根据时间戳删除旧的数据