Index: src/java/org/apache/nutch/fetcher/Fetcher.java =================================================================== --- src/java/org/apache/nutch/fetcher/Fetcher.java (revision 504810) +++ src/java/org/apache/nutch/fetcher/Fetcher.java (working copy) @@ -18,6 +18,9 @@ package org.apache.nutch.fetcher; import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; // Commons Logging imports import org.apache.commons.logging.Log; @@ -301,36 +304,50 @@ } } - Parse parse = null; + Map parseMap = null; if (parsing && status == CrawlDatum.STATUS_FETCH_SUCCESS) { - ParseStatus parseStatus; try { - parse = this.parseUtil.parse(content); - parseStatus = parse.getData().getStatus(); + parseMap = this.parseUtil.parse(content); } catch (Exception e) { - parseStatus = new ParseStatus(e); + // ignore } - if (!parseStatus.isSuccess()) { - if (LOG.isWarnEnabled()) { - LOG.warn("Error parsing: " + key + ": " + parseStatus); - } - parse = parseStatus.getEmptyParse(getConf()); - } + // Calculate page signature. For non-parsing fetchers this will // be done in ParseSegment - byte[] signature = SignatureFactory.getSignature(getConf()).calculate(content, parse); + byte[] signature = + SignatureFactory.getSignature(getConf()).calculate(content, parseMap); metadata.set(Nutch.SIGNATURE_KEY, StringUtil.toHexString(signature)); datum.setSignature(signature); - // Ensure segment name and score are in parseData metadata - parse.getData().getContentMeta().set(Nutch.SEGMENT_NAME_KEY, segmentName); - parse.getData().getContentMeta().set(Nutch.SIGNATURE_KEY, StringUtil.toHexString(signature)); - try { - scfilters.passScoreAfterParsing(key, content, parse); - } catch (Exception e) { - if (LOG.isWarnEnabled()) { - e.printStackTrace(LogUtil.getWarnStream(LOG)); - LOG.warn("Couldn't pass score, url " + key + " (" + e + ")"); + + for (Entry entry : parseMap.entrySet()) { + Text url = new Text(entry.getKey()); + Parse parse = entry.getValue(); + ParseStatus parseStatus = parse.getData().getStatus(); + Map singleEntryMap = new HashMap(); + + if (!parseStatus.isSuccess()) { + if (LOG.isWarnEnabled()) { + LOG.warn("Error parsing: " + key + ": " + parseStatus); + } + parse = parseStatus.getEmptyParse(getConf()); } + + // Calculate signature + singleEntryMap.put(entry.getKey(), parse); + byte[] entrySig = + SignatureFactory.getSignature(getConf()).calculate(content, singleEntryMap); + // Ensure segment name and score are in parseData metadata + parse.getData().getContentMeta().set(Nutch.SEGMENT_NAME_KEY, segmentName); + parse.getData().getContentMeta().set(Nutch.SIGNATURE_KEY, + StringUtil.toHexString(entrySig)); + try { + scfilters.passScoreAfterParsing(url, content, parse); + } catch (Exception e) { + if (LOG.isWarnEnabled()) { + e.printStackTrace(LogUtil.getWarnStream(LOG)); + LOG.warn("Couldn't pass score, url " + key + " (" + e + ")"); + } + } } } @@ -340,15 +357,21 @@ (key, new FetcherOutput(datum, storingContent ? content : null, - parse != null ? new ParseImpl(parse) : null)); + parseMap != null ? parseMap : null)); } catch (IOException e) { if (LOG.isFatalEnabled()) { e.printStackTrace(LogUtil.getFatalStream(LOG)); LOG.fatal("fetcher caught:"+e.toString()); } } - if (parse != null) return parse.getData().getStatus(); - else return null; + + if (parseMap != null && !parseMap.isEmpty()) { + ParseData pd = parseMap.values().iterator().next().getData(); + if (pd != null) { + return pd.getStatus(); + } + } + return null; } } Index: src/java/org/apache/nutch/fetcher/FetcherOutput.java =================================================================== --- src/java/org/apache/nutch/fetcher/FetcherOutput.java (revision 504810) +++ src/java/org/apache/nutch/fetcher/FetcherOutput.java (working copy) @@ -18,6 +18,9 @@ package org.apache.nutch.fetcher; import java.io.*; +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; import org.apache.hadoop.io.*; import org.apache.nutch.crawl.CrawlDatum; @@ -30,22 +33,28 @@ public final class FetcherOutput implements Writable, Configurable { private CrawlDatum crawlDatum; private Content content; - private ParseImpl parse; + private Map parseMap; private Configuration conf; public FetcherOutput() {} public FetcherOutput(CrawlDatum crawlDatum, Content content, - ParseImpl parse) { + Map parseMap) { this.crawlDatum = crawlDatum; this.content = content; - this.parse = parse; + this.parseMap = parseMap; } public final void readFields(DataInput in) throws IOException { this.crawlDatum = CrawlDatum.read(in); this.content = in.readBoolean() ? Content.read(in) : null; - this.parse = in.readBoolean() ? ParseImpl.read(in, this.conf) : null; + this.parseMap = new HashMap(); + int parseMapSize = in.readInt(); + for (int i = 0; i < parseMapSize; i++) { + String url = Text.readString(in); + Parse parse = ParseImpl.read(in, this.conf); + parseMap.put(url, parse); + } } public final void write(DataOutput out) throws IOException { @@ -56,15 +65,24 @@ content.write(out); } - out.writeBoolean(parse != null); - if (parse != null) { - parse.write(out); + if (parseMap == null) { + out.writeInt(0); + } else { + out.writeInt(parseMap.size()); + for (Entry entry : parseMap.entrySet()) { + String url = entry.getKey(); + ParseImpl parseImpl = new ParseImpl(entry.getValue()); + Text.writeString(out, url); + parseImpl.write(out); + } + } + } public CrawlDatum getCrawlDatum() { return crawlDatum; } public Content getContent() { return content; } - public ParseImpl getParse() { return parse; } + public Map getParse() { return parseMap; } public boolean equals(Object o) { if (!(o instanceof FetcherOutput)) Index: src/java/org/apache/nutch/fetcher/Fetcher2.java =================================================================== --- src/java/org/apache/nutch/fetcher/Fetcher2.java (revision 504810) +++ src/java/org/apache/nutch/fetcher/Fetcher2.java (working copy) @@ -21,6 +21,7 @@ import java.net.URL; import java.net.UnknownHostException; import java.util.*; +import java.util.Map.Entry; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@ -663,36 +664,50 @@ } } - Parse parse = null; + Map parseMap = null; if (parsing && status == CrawlDatum.STATUS_FETCH_SUCCESS) { - ParseStatus parseStatus; try { - parse = this.parseUtil.parse(content); - parseStatus = parse.getData().getStatus(); + parseMap = this.parseUtil.parse(content); } catch (Exception e) { - parseStatus = new ParseStatus(e); + // ignore } - if (!parseStatus.isSuccess()) { - if (LOG.isWarnEnabled()) { - LOG.warn("Error parsing: " + key + ": " + parseStatus); - } - parse = parseStatus.getEmptyParse(getConf()); - } + // Calculate page signature. For non-parsing fetchers this will // be done in ParseSegment - byte[] signature = SignatureFactory.getSignature(getConf()).calculate(content, parse); + byte[] signature = + SignatureFactory.getSignature(getConf()).calculate(content, parseMap); metadata.set(Nutch.SIGNATURE_KEY, StringUtil.toHexString(signature)); datum.setSignature(signature); - // Ensure segment name and score are in parseData metadata - parse.getData().getContentMeta().set(Nutch.SEGMENT_NAME_KEY, segmentName); - parse.getData().getContentMeta().set(Nutch.SIGNATURE_KEY, StringUtil.toHexString(signature)); - try { - scfilters.passScoreAfterParsing(key, content, parse); - } catch (Exception e) { - if (LOG.isWarnEnabled()) { - e.printStackTrace(LogUtil.getWarnStream(LOG)); - LOG.warn("Couldn't pass score, url " + key + " (" + e + ")"); + + for (Entry entry : parseMap.entrySet()) { + Text url = new Text(entry.getKey()); + Parse parse = entry.getValue(); + ParseStatus parseStatus = parse.getData().getStatus(); + Map singleEntryMap = new HashMap(); + + if (!parseStatus.isSuccess()) { + if (LOG.isWarnEnabled()) { + LOG.warn("Error parsing: " + key + ": " + parseStatus); + } + parse = parseStatus.getEmptyParse(getConf()); } + + // Calculate signature + singleEntryMap.put(entry.getKey(), parse); + byte[] entrySig = + SignatureFactory.getSignature(getConf()).calculate(content, singleEntryMap); + // Ensure segment name and score are in parseData metadata + parse.getData().getContentMeta().set(Nutch.SEGMENT_NAME_KEY, segmentName); + parse.getData().getContentMeta().set(Nutch.SIGNATURE_KEY, + StringUtil.toHexString(entrySig)); + try { + scfilters.passScoreAfterParsing(url, content, parse); + } catch (Exception e) { + if (LOG.isWarnEnabled()) { + e.printStackTrace(LogUtil.getWarnStream(LOG)); + LOG.warn("Couldn't pass score, url " + key + " (" + e + ")"); + } + } } } @@ -702,15 +717,21 @@ (key, new FetcherOutput(datum, storingContent ? content : null, - parse != null ? new ParseImpl(parse) : null)); + parseMap != null ? parseMap : null)); } catch (IOException e) { if (LOG.isFatalEnabled()) { e.printStackTrace(LogUtil.getFatalStream(LOG)); LOG.fatal("fetcher caught:"+e.toString()); } } - if (parse != null) return parse.getData().getStatus(); - else return null; + + if (parseMap != null && !parseMap.isEmpty()) { + ParseData pd = parseMap.values().iterator().next().getData(); + if (pd != null) { + return pd.getStatus(); + } + } + return null; } } Index: src/java/org/apache/nutch/fetcher/FetcherOutputFormat.java =================================================================== --- src/java/org/apache/nutch/fetcher/FetcherOutputFormat.java (revision 504810) +++ src/java/org/apache/nutch/fetcher/FetcherOutputFormat.java (working copy) @@ -18,6 +18,8 @@ package org.apache.nutch.fetcher; import java.io.IOException; +import java.util.Map; +import java.util.Map.Entry; import org.apache.nutch.crawl.CrawlDatum; import org.apache.hadoop.fs.FileSystem; @@ -34,6 +36,8 @@ import org.apache.hadoop.mapred.Reporter; import org.apache.hadoop.util.Progressable; +import org.apache.nutch.parse.Parse; +import org.apache.nutch.parse.ParseImpl; import org.apache.nutch.parse.ParseOutputFormat; import org.apache.nutch.protocol.Content; @@ -84,8 +88,12 @@ contentOut.append(key, fo.getContent()); } - if (fo.getParse() != null) { - parseOut.write(key, fo.getParse()); + Map parseMap = fo.getParse(); + if (parseMap != null) { + for (Entry entry : parseMap.entrySet()) { + parseOut.write(new Text(entry.getKey()), + new ParseImpl(entry.getValue())); + } } } Index: src/java/org/apache/nutch/crawl/Signature.java =================================================================== --- src/java/org/apache/nutch/crawl/Signature.java (revision 504810) +++ src/java/org/apache/nutch/crawl/Signature.java (working copy) @@ -17,6 +17,8 @@ package org.apache.nutch.crawl; +import java.util.Map; + import org.apache.nutch.parse.Parse; import org.apache.nutch.protocol.Content; import org.apache.hadoop.conf.Configuration; @@ -25,7 +27,7 @@ public abstract class Signature implements Configurable { protected Configuration conf; - public abstract byte[] calculate(Content content, Parse parse); + public abstract byte[] calculate(Content content, Map parseMap); public Configuration getConf() { return conf; Index: src/java/org/apache/nutch/crawl/MD5Signature.java =================================================================== --- src/java/org/apache/nutch/crawl/MD5Signature.java (revision 504810) +++ src/java/org/apache/nutch/crawl/MD5Signature.java (working copy) @@ -17,6 +17,8 @@ package org.apache.nutch.crawl; +import java.util.Map; + import org.apache.hadoop.io.MD5Hash; import org.apache.nutch.parse.Parse; import org.apache.nutch.protocol.Content; @@ -30,7 +32,7 @@ */ public class MD5Signature extends Signature { - public byte[] calculate(Content content, Parse parse) { + public byte[] calculate(Content content, Map parseMap) { byte[] data = content.getContent(); if (data == null) data = content.getUrl().getBytes(); return MD5Hash.digest(data).getDigest(); Index: src/java/org/apache/nutch/crawl/TextProfileSignature.java =================================================================== --- src/java/org/apache/nutch/crawl/TextProfileSignature.java (revision 504810) +++ src/java/org/apache/nutch/crawl/TextProfileSignature.java (working copy) @@ -26,6 +26,7 @@ import java.util.Comparator; import java.util.HashMap; import java.util.Iterator; +import java.util.Map; import org.apache.hadoop.io.MD5Hash; import org.apache.nutch.parse.Parse; @@ -63,13 +64,19 @@ Signature fallback = new MD5Signature(); - public byte[] calculate(Content content, Parse parse) { + public byte[] calculate(Content content, Map parseMap) { int MIN_TOKEN_LEN = getConf().getInt("db.signature.text_profile.min_token_len", 2); float QUANT_RATE = getConf().getFloat("db.signature.text_profile.quant_rate", 0.01f); HashMap tokens = new HashMap(); String text = null; - if (parse != null) text = parse.getText(); - if (text == null || text.length() == 0) return fallback.calculate(content, parse); + if (parseMap != null) { + StringBuffer textBuf = new StringBuffer(); + for (Parse parse : parseMap.values()) { + textBuf.append(parse.getText()); + } + text = textBuf.toString(); + } + if (text == null || text.length() == 0) return fallback.calculate(content, parseMap); StringBuffer curToken = new StringBuffer(); int maxFreq = 0; for (int i = 0; i < text.length(); i++) { @@ -171,7 +178,9 @@ text.append(line); } br.close(); - byte[] signature = sig.calculate(null, new ParseImpl(text.toString(), null)); + Map parseMap = new HashMap(); + parseMap.put("", new ParseImpl(text.toString(), null)); + byte[] signature = sig.calculate(null, parseMap); res.put(files[i].toString(), signature); } Iterator it = res.keySet().iterator(); Index: src/java/org/apache/nutch/parse/ParseUtil.java =================================================================== --- src/java/org/apache/nutch/parse/ParseUtil.java (revision 504810) +++ src/java/org/apache/nutch/parse/ParseUtil.java (working copy) @@ -17,6 +17,9 @@ package org.apache.nutch.parse; // Commons Logging imports +import java.util.HashMap; +import java.util.Map; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -59,10 +62,10 @@ * WARNING level, and an empty parse is returned. * * @param content The content to try and parse. - * @return A {@link Parse} object containing the parsed data. + * @return <key, {@link Parse}> pairs. * @throws ParseException If no suitable parser is found to perform the parse. */ - public Parse parse(Content content) throws ParseException { + public Map parse(Content content) throws ParseException { Parser[] parsers = null; try { @@ -76,25 +79,17 @@ throw new ParseException(e.getMessage()); } - Parse parse = null; + Map parseMap = null; for (int i=0; iextId, i.e., the Parser's extension ID. * If a suitable {@link Parser} is not found, then a WARNING * level message is logged, and a ParseException is thrown. If the parse is - * uncessful for any other reason, then a WARNING level + * unsuccessful for any other reason, then a WARNING level * message is logged, and a ParseStatus.getEmptyParse() is * returned. * @@ -110,15 +105,14 @@ * to parse the specified content. * @param content The content to parse. * - * @return A {@link Parse} object if the parse is successful, otherwise, - * a ParseStatus.getEmptyParse(). + * @return <key, {@link Parse}> pairs if the parse is successful, otherwise, + * a single <key, ParseStatus.getEmptyParse()> pair. * * @throws ParseException If there is no suitable {@link Parser} found * to perform the parse. */ - public Parse parseByExtensionId(String extId, Content content) + public Map parseByExtensionId(String extId, Content content) throws ParseException { - Parse parse = null; Parser p = null; try { @@ -131,17 +125,48 @@ throw new ParseException(e.getMessage()); } - parse = p.getParse(content); - - if (parse != null && parse.getData().getStatus().isSuccess()) { - return parse; + Map parseMap = p.getParse(content); + if (processParseMap(content, parseMap)) { + return parseMap; } else { - if (LOG.isWarnEnabled()) { - LOG.warn("Unable to successfully parse content " + content.getUrl() + - " of type " + content.getContentType()); + return getEmptyParse(content, parseMap); + } + } + + private boolean processParseMap(Content content, + Map parseMap) { + if (parseMap != null) { + for (String key : parseMap.keySet()) { + Parse parse = parseMap.get(key); + if (parse == null || !parse.getData().getStatus().isSuccess()) + parseMap.remove(key); } - return new ParseStatus().getEmptyParse(this.conf); + if (!parseMap.isEmpty()) + return true; } - } + + return false; + } + private Map getEmptyParse(Content content, + Map parseMap) { + if (LOG.isWarnEnabled()) { + LOG.warn("Unable to successfully parse content " + content.getUrl() + + " of type " + content.getContentType()); + } + + if (parseMap != null && !parseMap.isEmpty()) { + ParseData pd = parseMap.values().iterator().next().getData(); + if (pd != null) { + parseMap.clear(); + parseMap.put(content.getUrl(), pd.getStatus().getEmptyParse(conf)); + } + } else { + parseMap = new HashMap(); + parseMap.put(content.getUrl(), new ParseStatus().getEmptyParse(conf)); + } + + return parseMap; + } + } Index: src/java/org/apache/nutch/parse/ParserChecker.java =================================================================== --- src/java/org/apache/nutch/parse/ParserChecker.java (revision 504810) +++ src/java/org/apache/nutch/parse/ParserChecker.java (working copy) @@ -17,6 +17,9 @@ package org.apache.nutch.parse; +import java.util.Map; +import java.util.Map.Entry; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -94,13 +97,18 @@ LOG.info("contentType: "+contentType); } - Parse parse = new ParseUtil(conf).parse(content); + Map parseMap = new ParseUtil(conf).parse(content); - System.out.print("---------\nParseData\n---------\n"); - System.out.print(parse.getData().toString()); - if (dumpText) { - System.out.print("---------\nParseText\n---------\n"); - System.out.print(parse.getText()); + for (Entry entry : parseMap.entrySet()) { + Parse parse = entry.getValue(); + System.out.print("---------\nUrl\n---------------\n"); + System.out.print(entry.getKey()); + System.out.print("---------\nParseData\n---------\n"); + System.out.print(parse.getData().toString()); + if (dumpText) { + System.out.print("---------\nParseText\n---------\n"); + System.out.print(parse.getText()); + } } System.exit(0); Index: src/java/org/apache/nutch/parse/Parser.java =================================================================== --- src/java/org/apache/nutch/parse/Parser.java (revision 504810) +++ src/java/org/apache/nutch/parse/Parser.java (working copy) @@ -18,6 +18,8 @@ package org.apache.nutch.parse; // Hadoop imports +import java.util.Map; + import org.apache.hadoop.conf.Configurable; // Nutch imports @@ -33,5 +35,5 @@ public final static String X_POINT_ID = Parser.class.getName(); /** Creates the parse for some content. */ - Parse getParse(Content c); + Map getParse(Content c); } Index: src/java/org/apache/nutch/parse/ParseSegment.java =================================================================== --- src/java/org/apache/nutch/parse/ParseSegment.java (revision 504810) +++ src/java/org/apache/nutch/parse/ParseSegment.java (working copy) @@ -34,6 +34,7 @@ import java.io.*; import java.util.*; +import java.util.Map.Entry; /* Parse content in a segment. */ public class ParseSegment extends Configured implements Mapper, Reducer { @@ -70,32 +71,35 @@ Content content = (Content) value; content.forceInflate(); - Parse parse = null; - ParseStatus status; + Map parseMap = null; try { - parse = new ParseUtil(getConf()).parse(content); - status = parse.getData().getStatus(); + parseMap = new ParseUtil(getConf()).parse(content); } catch (Exception e) { - status = new ParseStatus(e); + // ignore } - - // compute the new signature - byte[] signature = SignatureFactory.getSignature(getConf()).calculate(content, parse); + + byte[] signature = SignatureFactory.getSignature(getConf()).calculate(content, parseMap); content.getMetadata().set(Nutch.SIGNATURE_KEY, StringUtil.toHexString(signature)); - if (status.isSuccess()) { - try { - scfilters.passScoreAfterParsing((Text)key, content, parse); - } catch (ScoringFilterException e) { - if (LOG.isWarnEnabled()) { - e.printStackTrace(LogUtil.getWarnStream(LOG)); - LOG.warn("Error passing score: "+key+": "+e.getMessage()); + for (Entry entry : parseMap.entrySet()) { + Text url = new Text(entry.getKey()); + Parse parse = entry.getValue(); + ParseStatus status = parse.getData().getStatus(); + // compute the new signature + if (status.isSuccess()) { + try { + scfilters.passScoreAfterParsing(url, content, parse); + } catch (ScoringFilterException e) { + if (LOG.isWarnEnabled()) { + e.printStackTrace(LogUtil.getWarnStream(LOG)); + LOG.warn("Error passing score: "+ url +": "+e.getMessage()); + } + return; } - return; + output.collect(url, new ParseImpl(parse.getText(), parse.getData())); + } else if (LOG.isWarnEnabled()) { + LOG.warn("Error parsing: " + url + ": "+status.toString()); } - output.collect(key, new ParseImpl(parse.getText(), parse.getData())); - } else if (LOG.isWarnEnabled()) { - LOG.warn("Error parsing: "+key+": "+status.toString()); } }