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();