[
https://issues.apache.org/jira/browse/NUTCH-2455?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18136265#comment-18136265
]
ASF GitHub Bot commented on NUTCH-2455:
---------------------------------------
lewismc commented on code in PR #888:
URL: https://github.com/apache/nutch/pull/888#discussion_r4438570485
##########
src/java/org/apache/nutch/crawl/Generator.java:
##########
@@ -148,35 +208,320 @@ public String toString() {
}
}
- /** Selects entries due for fetch. */
+ /**
+ * Composite key for secondary sorting. Contains score and hostname.
+ * Used to ensure HostDb entries arrive before CrawlDb entries in the
reducer.
+ *
+ * For HostDb entries: FloatTextPair(-Float.MAX_VALUE, hostname)
+ * For CrawlDb entries: FloatTextPair(score, "")
+ */
+ public static class FloatTextPair implements
WritableComparable<FloatTextPair> {
+ public FloatWritable first;
+ public Text second;
+
+ public FloatTextPair() {
+ this.first = new FloatWritable();
+ this.second = new Text();
+ }
+
+ public FloatTextPair(FloatWritable first, Text second) {
+ this.first = first;
+ this.second = second;
+ }
+
+ public FloatTextPair(float first, String second) {
+ this.first = new FloatWritable(first);
+ this.second = new Text(second);
+ }
+
+ public FloatWritable getFirst() {
+ return first;
+ }
+
+ public void setFirst(FloatWritable first) {
+ this.first = first;
+ }
+
+ public Text getSecond() {
+ return second;
+ }
+
+ public void setSecond(Text second) {
+ this.second = second;
+ }
+
+ public void set(FloatWritable first, Text second) {
+ this.first = first;
+ this.second = second;
+ }
+
+ @Override
+ public int hashCode() {
+ return first.hashCode() * 163 + second.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof FloatTextPair) {
+ FloatTextPair tp = (FloatTextPair) obj;
+ return first.equals(tp.getFirst()) && second.equals(tp.getSecond());
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return first + "\t" + second;
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ first.readFields(in);
+ second.readFields(in);
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ first.write(out);
+ second.write(out);
+ }
+
+ @Override
+ public int compareTo(FloatTextPair tp) {
+ int cmp = first.compareTo(tp.getFirst());
+ if (cmp != 0)
+ return cmp;
+ return second.compareTo(tp.getSecond());
+ }
+ }
+
+ /**
+ * Comparator that ensures HostDb entries (with hostname in second field)
+ * sort before CrawlDb entries (with empty second field) for the same host.
+ * This enables the secondary sorting pattern for NUTCH-2455.
+ */
+ public static class ScoreHostKeyComparator extends WritableComparator {
+ protected ScoreHostKeyComparator() {
+ super(FloatTextPair.class, true);
+ }
+
+ @Override
+ public int compare(WritableComparable w1, WritableComparable w2) {
+ FloatTextPair key1 = (FloatTextPair) w1;
+ FloatTextPair key2 = (FloatTextPair) w2;
+
+ boolean isKey1HostDatum = key1.second.getLength() > 0;
+ boolean isKey2HostDatum = key2.second.getLength() > 0;
+
+ if (isKey1HostDatum && isKey2HostDatum) {
+ // Both are HostDb entries, sort by hostname
+ return key1.second.compareTo(key2.second);
+ } else {
+ if (isKey1HostDatum == isKey2HostDatum) {
+ // Both are CrawlDb entries, sort by score descending
+ return -1 * key1.first.compareTo(key2.first);
+ } else if (isKey1HostDatum) {
+ // HostDb entries come before CrawlDb entries
+ return -1;
+ } else {
+ return 1;
+ }
+ }
+ }
+ }
+
+ /**
+ * Mapper that reads HostDb and emits entries for secondary sorting.
+ * Uses a very low score (-Float.MAX_VALUE) to ensure HostDb entries
+ * sort before CrawlDb entries.
+ */
+ public static class HostDbReaderMapper
+ extends Mapper<Text, HostDatum, FloatTextPair, SelectorEntry> {
+
+ @Override
+ public void map(Text hostname, HostDatum value, Context context)
+ throws IOException, InterruptedException {
+ SelectorEntry hostDataSelector = new SelectorEntry();
+ hostDataSelector.hasHostDatum.set(true);
+ try {
+ hostDataSelector.hostdatum = (HostDatum) value.clone();
+ } catch (CloneNotSupportedException e) {
+ hostDataSelector.hostdatum = value;
+ }
+
+ // Use very low score and hostname to ensure HostDb entries
+ // sort before CrawlDb entries for the same host
+ context.write(new FloatTextPair(new FloatWritable(-Float.MAX_VALUE),
+ hostname), hostDataSelector);
+ }
+ }
+
+ /** Selects entries due for fetch. Partitions by host/domain/IP (with HostDb
support). */
+ public static class SelectorWithHostDb extends Partitioner<FloatTextPair,
Writable>
+ implements Configurable {
+
+ private final URLPartitioner partitioner = new URLPartitioner();
+ private final Map<String, Integer> partitionByHost = new HashMap<>();
+ private Configuration conf;
+ private int seed;
+ private int cachedReduceTasks = -1;
+ private String mode = URLPartitioner.PARTITION_MODE_HOST;
+
+ /**
+ * Partition by host / domain or IP.
+ * For HostDb entries (key.second is non-empty), partition by the hostname.
+ * For CrawlDb entries (key.second is empty), use the URLPartitioner.
+ *
+ * Note: When partition.url.mode is not "byHost", HostDb entries may not
+ * be correctly grouped with their corresponding CrawlDb entries. For best
+ * results, use "byHost" mode (the default) when using HostDb.
+ */
+ @Override
+ public int getPartition(FloatTextPair key, Writable value,
+ int numReduceTasks) {
+ SelectorEntry entry = (SelectorEntry) value;
+ // For HostDb entries, use the hostname from the key
+ // For CrawlDb entries, use the URL from the entry
+ if (key.second.getLength() > 0) {
+ // HostDb entry - partition by hostname
+ // This works best with "byHost" mode; other modes may cause
+ // HostDb entries to not be grouped with their CrawlDb entries
+ int hashCode = key.second.toString().hashCode();
+ hashCode ^= seed;
+ return (hashCode & Integer.MAX_VALUE) % numReduceTasks;
+ }
+ if (cachedReduceTasks != numReduceTasks) {
+ partitionByHost.clear();
+ cachedReduceTasks = numReduceTasks;
+ }
+ String host = rawHost(entry.url);
+ if (host != null) {
+ Integer cached = partitionByHost.get(host);
+ if (cached != null) {
+ return cached;
+ }
+ }
+ int part = partitioner.getPartition(entry.url, key.first,
numReduceTasks);
+ if (host != null) {
+ partitionByHost.put(host, part);
+ }
+ return part;
+ }
+
+ /**
+ * Host used as the CrawlDb partition cache key. Returns null when the URL
+ * has no scheme or host, so those URLs keep the uncached partitioner path.
+ */
+ private static String rawHost(Text url) {
+ int length = url.getLength();
+ if (length <= 0) {
+ return null;
+ }
+ byte[] bytes = url.getBytes();
+ int scheme = -1;
+ for (int i = 0; i + 2 < length; i++) {
+ if (bytes[i] == ':' && bytes[i + 1] == '/' && bytes[i + 2] == '/') {
+ scheme = i;
+ break;
+ }
+ }
+ if (scheme < 0) {
+ return null;
+ }
+ int start = scheme + 3;
+ int authorityEnd = start;
+ int at = -1;
+ for (int i = start; i < length; i++) {
+ byte c = bytes[i];
+ if (c == '/' || c == '?' || c == '#') {
+ break;
+ }
+ if (c == '@') {
+ at = i;
+ }
+ authorityEnd = i + 1;
+ }
+ int hostStart = at >= 0 ? at + 1 : start;
+ int hostEnd;
+ if (hostStart < authorityEnd && bytes[hostStart] == '[') {
+ hostEnd = hostStart + 1;
+ while (hostEnd < authorityEnd && bytes[hostEnd] != ']') {
+ hostEnd++;
+ }
+ if (hostEnd < authorityEnd && bytes[hostEnd] == ']') {
+ hostEnd++;
+ }
+ } else {
+ hostEnd = hostStart;
+ while (hostEnd < authorityEnd && bytes[hostEnd] != ':') {
+ hostEnd++;
+ }
+ }
+ if (hostEnd <= hostStart) {
+ return null;
+ }
+ return new String(bytes, hostStart, hostEnd - hostStart,
+ StandardCharsets.UTF_8).toLowerCase(Locale.ROOT);
+ }
+
+ @Override
+ public Configuration getConf() {
+ return conf;
+ }
+
+ @Override
+ public void setConf(Configuration conf) {
+ this.conf = conf;
+ this.seed = conf.getInt("partition.url.seed", 0);
+ this.mode = conf.get(URLPartitioner.PARTITION_MODE_KEY,
+ URLPartitioner.PARTITION_MODE_HOST);
+ partitionByHost.clear();
+ cachedReduceTasks = -1;
+ partitioner.setConf(conf);
+
+ // Warn if mode is not byHost and hostdb is configured
+ String hostdb = conf.get(GENERATOR_HOSTDB);
+ if (hostdb != null && !mode.equals(URLPartitioner.PARTITION_MODE_HOST)) {
+ LOG.warn("HostDb is configured but partition.url.mode is '{}'. " +
Review Comment:
Done. When `generate.hostdb` is set and `partition.url.mode` is not
`byHost`, HostDb generation logs an error with the configured mode and forces
`byHost` for this partitioner. The job configuration is left unchanged, so
fetch and `generate.count.mode` keep the configured mode.
Forcing `byHost` is what lets the HostDb row and its CrawlDb URLs share a
reducer. Both now hash the lowercased host through
`URLPartitioner.getPartitionForRoot`.
##########
src/java/org/apache/nutch/crawl/URLPartitioner.java:
##########
@@ -71,6 +71,33 @@ public Configuration getConf() {
return conf;
}
+ /**
+ * Returns the normalized URL root (host, domain, or IP) for the given URL
key.
+ * This is used for partitioning URLs by their root component.
+ *
+ * @param key
+ * The URL as a Text object.
+ * @return The normalized URL root based on the partition mode.
+ */
+ public String getNormalizedURLRoot(Text key) {
Review Comment:
No. Nothing calls `getNormalizedURLRoot`. I removed it.
##########
src/java/org/apache/nutch/crawl/URLPartitioner.java:
##########
@@ -71,6 +71,33 @@ public Configuration getConf() {
return conf;
}
+ /**
+ * Returns the normalized URL root (host, domain, or IP) for the given URL
key.
+ * This is used for partitioning URLs by their root component.
+ *
+ * @param key
+ * The URL as a Text object.
+ * @return The normalized URL root based on the partition mode.
+ */
+ public String getNormalizedURLRoot(Text key) {
+ String urlString = key.toString();
+ URL url = null;
+ try {
+ urlString = normalizers.normalize(urlString,
URLNormalizers.SCOPE_PARTITION);
+ url = new URL(urlString);
+ } catch (MalformedURLException e) {
+ LOG.warn("Malformed URL: '{}'", urlString);
+ return urlString;
+ }
+
+ try {
+ urlString = URLUtil.getUrlRootByMode(url, mode);
Review Comment:
Agreed. The reducers already call `URLUtil.getUrlRootByMode` directly.
`getPartition` now does the same after `SCOPE_PARTITION` normalization, and the
wrapper method is gone.
##########
src/java/org/apache/nutch/crawl/URLPartitioner.java:
##########
Review Comment:
Done. After normalization and parsing, `getPartition` takes
`URLUtil.getUrlRootByMode(url, mode)` and hashes it with `getPartitionForRoot`.
A malformed URL still hashes the string it already has. A `byIP` lookup that
throws `UnknownHostException` still logs and hashes 0.
##########
src/java/org/apache/nutch/crawl/Generator.java:
##########
@@ -148,35 +208,320 @@ public String toString() {
}
}
- /** Selects entries due for fetch. */
+ /**
+ * Composite key for secondary sorting. Contains score and hostname.
+ * Used to ensure HostDb entries arrive before CrawlDb entries in the
reducer.
+ *
+ * For HostDb entries: FloatTextPair(-Float.MAX_VALUE, hostname)
+ * For CrawlDb entries: FloatTextPair(score, "")
+ */
+ public static class FloatTextPair implements
WritableComparable<FloatTextPair> {
+ public FloatWritable first;
+ public Text second;
+
+ public FloatTextPair() {
+ this.first = new FloatWritable();
+ this.second = new Text();
+ }
+
+ public FloatTextPair(FloatWritable first, Text second) {
+ this.first = first;
+ this.second = second;
+ }
+
+ public FloatTextPair(float first, String second) {
+ this.first = new FloatWritable(first);
+ this.second = new Text(second);
+ }
+
+ public FloatWritable getFirst() {
+ return first;
+ }
+
+ public void setFirst(FloatWritable first) {
+ this.first = first;
+ }
+
+ public Text getSecond() {
+ return second;
+ }
+
+ public void setSecond(Text second) {
+ this.second = second;
+ }
+
+ public void set(FloatWritable first, Text second) {
+ this.first = first;
+ this.second = second;
+ }
+
+ @Override
+ public int hashCode() {
+ return first.hashCode() * 163 + second.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof FloatTextPair) {
+ FloatTextPair tp = (FloatTextPair) obj;
+ return first.equals(tp.getFirst()) && second.equals(tp.getSecond());
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return first + "\t" + second;
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ first.readFields(in);
+ second.readFields(in);
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ first.write(out);
+ second.write(out);
+ }
+
+ @Override
+ public int compareTo(FloatTextPair tp) {
+ int cmp = first.compareTo(tp.getFirst());
+ if (cmp != 0)
+ return cmp;
+ return second.compareTo(tp.getSecond());
+ }
+ }
+
+ /**
+ * Comparator that ensures HostDb entries (with hostname in second field)
+ * sort before CrawlDb entries (with empty second field) for the same host.
+ * This enables the secondary sorting pattern for NUTCH-2455.
+ */
+ public static class ScoreHostKeyComparator extends WritableComparator {
+ protected ScoreHostKeyComparator() {
+ super(FloatTextPair.class, true);
+ }
+
+ @Override
+ public int compare(WritableComparable w1, WritableComparable w2) {
+ FloatTextPair key1 = (FloatTextPair) w1;
+ FloatTextPair key2 = (FloatTextPair) w2;
+
+ boolean isKey1HostDatum = key1.second.getLength() > 0;
+ boolean isKey2HostDatum = key2.second.getLength() > 0;
+
+ if (isKey1HostDatum && isKey2HostDatum) {
+ // Both are HostDb entries, sort by hostname
+ return key1.second.compareTo(key2.second);
+ } else {
+ if (isKey1HostDatum == isKey2HostDatum) {
+ // Both are CrawlDb entries, sort by score descending
+ return -1 * key1.first.compareTo(key2.first);
+ } else if (isKey1HostDatum) {
+ // HostDb entries come before CrawlDb entries
+ return -1;
+ } else {
+ return 1;
+ }
+ }
+ }
+ }
+
+ /**
+ * Mapper that reads HostDb and emits entries for secondary sorting.
+ * Uses a very low score (-Float.MAX_VALUE) to ensure HostDb entries
+ * sort before CrawlDb entries.
+ */
+ public static class HostDbReaderMapper
+ extends Mapper<Text, HostDatum, FloatTextPair, SelectorEntry> {
+
+ @Override
+ public void map(Text hostname, HostDatum value, Context context)
+ throws IOException, InterruptedException {
+ SelectorEntry hostDataSelector = new SelectorEntry();
+ hostDataSelector.hasHostDatum.set(true);
+ try {
+ hostDataSelector.hostdatum = (HostDatum) value.clone();
+ } catch (CloneNotSupportedException e) {
+ hostDataSelector.hostdatum = value;
+ }
+
+ // Use very low score and hostname to ensure HostDb entries
+ // sort before CrawlDb entries for the same host
+ context.write(new FloatTextPair(new FloatWritable(-Float.MAX_VALUE),
+ hostname), hostDataSelector);
+ }
+ }
+
+ /** Selects entries due for fetch. Partitions by host/domain/IP (with HostDb
support). */
+ public static class SelectorWithHostDb extends Partitioner<FloatTextPair,
Writable>
+ implements Configurable {
+
+ private final URLPartitioner partitioner = new URLPartitioner();
+ private final Map<String, Integer> partitionByHost = new HashMap<>();
+ private Configuration conf;
+ private int seed;
+ private int cachedReduceTasks = -1;
+ private String mode = URLPartitioner.PARTITION_MODE_HOST;
+
+ /**
+ * Partition by host / domain or IP.
+ * For HostDb entries (key.second is non-empty), partition by the hostname.
+ * For CrawlDb entries (key.second is empty), use the URLPartitioner.
+ *
+ * Note: When partition.url.mode is not "byHost", HostDb entries may not
+ * be correctly grouped with their corresponding CrawlDb entries. For best
+ * results, use "byHost" mode (the default) when using HostDb.
+ */
+ @Override
+ public int getPartition(FloatTextPair key, Writable value,
+ int numReduceTasks) {
+ SelectorEntry entry = (SelectorEntry) value;
+ // For HostDb entries, use the hostname from the key
+ // For CrawlDb entries, use the URL from the entry
+ if (key.second.getLength() > 0) {
+ // HostDb entry - partition by hostname
+ // This works best with "byHost" mode; other modes may cause
+ // HostDb entries to not be grouped with their CrawlDb entries
+ int hashCode = key.second.toString().hashCode();
+ hashCode ^= seed;
Review Comment:
Done. That hash now lives on `URLPartitioner.getPartitionForRoot`. The
HostDb row calls it with the hostname lowercased, using the same
`partition.url.seed` as the CrawlDb rows.
##########
src/java/org/apache/nutch/crawl/Generator.java:
##########
@@ -148,35 +208,320 @@ public String toString() {
}
}
- /** Selects entries due for fetch. */
+ /**
+ * Composite key for secondary sorting. Contains score and hostname.
+ * Used to ensure HostDb entries arrive before CrawlDb entries in the
reducer.
+ *
+ * For HostDb entries: FloatTextPair(-Float.MAX_VALUE, hostname)
+ * For CrawlDb entries: FloatTextPair(score, "")
+ */
+ public static class FloatTextPair implements
WritableComparable<FloatTextPair> {
+ public FloatWritable first;
+ public Text second;
+
+ public FloatTextPair() {
+ this.first = new FloatWritable();
+ this.second = new Text();
+ }
+
+ public FloatTextPair(FloatWritable first, Text second) {
+ this.first = first;
+ this.second = second;
+ }
+
+ public FloatTextPair(float first, String second) {
+ this.first = new FloatWritable(first);
+ this.second = new Text(second);
+ }
+
+ public FloatWritable getFirst() {
+ return first;
+ }
+
+ public void setFirst(FloatWritable first) {
+ this.first = first;
+ }
+
+ public Text getSecond() {
+ return second;
+ }
+
+ public void setSecond(Text second) {
+ this.second = second;
+ }
+
+ public void set(FloatWritable first, Text second) {
+ this.first = first;
+ this.second = second;
+ }
+
+ @Override
+ public int hashCode() {
+ return first.hashCode() * 163 + second.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof FloatTextPair) {
+ FloatTextPair tp = (FloatTextPair) obj;
+ return first.equals(tp.getFirst()) && second.equals(tp.getSecond());
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return first + "\t" + second;
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ first.readFields(in);
+ second.readFields(in);
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ first.write(out);
+ second.write(out);
+ }
+
+ @Override
+ public int compareTo(FloatTextPair tp) {
+ int cmp = first.compareTo(tp.getFirst());
+ if (cmp != 0)
+ return cmp;
+ return second.compareTo(tp.getSecond());
+ }
+ }
+
+ /**
+ * Comparator that ensures HostDb entries (with hostname in second field)
+ * sort before CrawlDb entries (with empty second field) for the same host.
+ * This enables the secondary sorting pattern for NUTCH-2455.
+ */
+ public static class ScoreHostKeyComparator extends WritableComparator {
+ protected ScoreHostKeyComparator() {
+ super(FloatTextPair.class, true);
+ }
+
+ @Override
+ public int compare(WritableComparable w1, WritableComparable w2) {
+ FloatTextPair key1 = (FloatTextPair) w1;
+ FloatTextPair key2 = (FloatTextPair) w2;
+
+ boolean isKey1HostDatum = key1.second.getLength() > 0;
+ boolean isKey2HostDatum = key2.second.getLength() > 0;
+
+ if (isKey1HostDatum && isKey2HostDatum) {
+ // Both are HostDb entries, sort by hostname
+ return key1.second.compareTo(key2.second);
+ } else {
+ if (isKey1HostDatum == isKey2HostDatum) {
+ // Both are CrawlDb entries, sort by score descending
+ return -1 * key1.first.compareTo(key2.first);
+ } else if (isKey1HostDatum) {
+ // HostDb entries come before CrawlDb entries
+ return -1;
+ } else {
+ return 1;
+ }
+ }
+ }
+ }
+
+ /**
+ * Mapper that reads HostDb and emits entries for secondary sorting.
+ * Uses a very low score (-Float.MAX_VALUE) to ensure HostDb entries
+ * sort before CrawlDb entries.
+ */
+ public static class HostDbReaderMapper
+ extends Mapper<Text, HostDatum, FloatTextPair, SelectorEntry> {
+
+ @Override
+ public void map(Text hostname, HostDatum value, Context context)
+ throws IOException, InterruptedException {
+ SelectorEntry hostDataSelector = new SelectorEntry();
+ hostDataSelector.hasHostDatum.set(true);
+ try {
+ hostDataSelector.hostdatum = (HostDatum) value.clone();
+ } catch (CloneNotSupportedException e) {
+ hostDataSelector.hostdatum = value;
+ }
+
+ // Use very low score and hostname to ensure HostDb entries
+ // sort before CrawlDb entries for the same host
+ context.write(new FloatTextPair(new FloatWritable(-Float.MAX_VALUE),
+ hostname), hostDataSelector);
+ }
+ }
+
+ /** Selects entries due for fetch. Partitions by host/domain/IP (with HostDb
support). */
+ public static class SelectorWithHostDb extends Partitioner<FloatTextPair,
Writable>
+ implements Configurable {
+
+ private final URLPartitioner partitioner = new URLPartitioner();
+ private final Map<String, Integer> partitionByHost = new HashMap<>();
+ private Configuration conf;
+ private int seed;
+ private int cachedReduceTasks = -1;
+ private String mode = URLPartitioner.PARTITION_MODE_HOST;
+
+ /**
+ * Partition by host / domain or IP.
+ * For HostDb entries (key.second is non-empty), partition by the hostname.
+ * For CrawlDb entries (key.second is empty), use the URLPartitioner.
+ *
+ * Note: When partition.url.mode is not "byHost", HostDb entries may not
+ * be correctly grouped with their corresponding CrawlDb entries. For best
+ * results, use "byHost" mode (the default) when using HostDb.
+ */
+ @Override
+ public int getPartition(FloatTextPair key, Writable value,
+ int numReduceTasks) {
+ SelectorEntry entry = (SelectorEntry) value;
+ // For HostDb entries, use the hostname from the key
+ // For CrawlDb entries, use the URL from the entry
+ if (key.second.getLength() > 0) {
+ // HostDb entry - partition by hostname
+ // This works best with "byHost" mode; other modes may cause
+ // HostDb entries to not be grouped with their CrawlDb entries
+ int hashCode = key.second.toString().hashCode();
+ hashCode ^= seed;
+ return (hashCode & Integer.MAX_VALUE) % numReduceTasks;
+ }
+ if (cachedReduceTasks != numReduceTasks) {
+ partitionByHost.clear();
+ cachedReduceTasks = numReduceTasks;
+ }
+ String host = rawHost(entry.url);
+ if (host != null) {
+ Integer cached = partitionByHost.get(host);
+ if (cached != null) {
+ return cached;
+ }
+ }
+ int part = partitioner.getPartition(entry.url, key.first,
numReduceTasks);
Review Comment:
Done, with the method named `getPartitionForRoot`. `byHost` is now
guaranteed for HostDb generation, and those CrawlDb rows do not run
`SCOPE_PARTITION`. `rawHost` already has the host, and
`SelectorWithHostDb.getPartition` passes it to `getPartitionForRoot`. A URL
with no scheme or host still falls through to `URLPartitioner.getPartition`.
##########
src/java/org/apache/nutch/crawl/Generator.java:
##########
@@ -148,35 +208,320 @@ public String toString() {
}
}
- /** Selects entries due for fetch. */
+ /**
+ * Composite key for secondary sorting. Contains score and hostname.
+ * Used to ensure HostDb entries arrive before CrawlDb entries in the
reducer.
+ *
+ * For HostDb entries: FloatTextPair(-Float.MAX_VALUE, hostname)
+ * For CrawlDb entries: FloatTextPair(score, "")
+ */
+ public static class FloatTextPair implements
WritableComparable<FloatTextPair> {
+ public FloatWritable first;
+ public Text second;
+
+ public FloatTextPair() {
+ this.first = new FloatWritable();
+ this.second = new Text();
+ }
+
+ public FloatTextPair(FloatWritable first, Text second) {
+ this.first = first;
+ this.second = second;
+ }
+
+ public FloatTextPair(float first, String second) {
+ this.first = new FloatWritable(first);
+ this.second = new Text(second);
+ }
+
+ public FloatWritable getFirst() {
+ return first;
+ }
+
+ public void setFirst(FloatWritable first) {
+ this.first = first;
+ }
+
+ public Text getSecond() {
+ return second;
+ }
+
+ public void setSecond(Text second) {
+ this.second = second;
+ }
+
+ public void set(FloatWritable first, Text second) {
+ this.first = first;
+ this.second = second;
+ }
+
+ @Override
+ public int hashCode() {
+ return first.hashCode() * 163 + second.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof FloatTextPair) {
+ FloatTextPair tp = (FloatTextPair) obj;
+ return first.equals(tp.getFirst()) && second.equals(tp.getSecond());
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return first + "\t" + second;
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ first.readFields(in);
+ second.readFields(in);
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ first.write(out);
+ second.write(out);
+ }
+
+ @Override
+ public int compareTo(FloatTextPair tp) {
+ int cmp = first.compareTo(tp.getFirst());
+ if (cmp != 0)
+ return cmp;
+ return second.compareTo(tp.getSecond());
+ }
+ }
+
+ /**
+ * Comparator that ensures HostDb entries (with hostname in second field)
+ * sort before CrawlDb entries (with empty second field) for the same host.
+ * This enables the secondary sorting pattern for NUTCH-2455.
+ */
+ public static class ScoreHostKeyComparator extends WritableComparator {
+ protected ScoreHostKeyComparator() {
+ super(FloatTextPair.class, true);
+ }
+
+ @Override
+ public int compare(WritableComparable w1, WritableComparable w2) {
+ FloatTextPair key1 = (FloatTextPair) w1;
+ FloatTextPair key2 = (FloatTextPair) w2;
+
+ boolean isKey1HostDatum = key1.second.getLength() > 0;
+ boolean isKey2HostDatum = key2.second.getLength() > 0;
+
+ if (isKey1HostDatum && isKey2HostDatum) {
+ // Both are HostDb entries, sort by hostname
+ return key1.second.compareTo(key2.second);
+ } else {
+ if (isKey1HostDatum == isKey2HostDatum) {
+ // Both are CrawlDb entries, sort by score descending
+ return -1 * key1.first.compareTo(key2.first);
+ } else if (isKey1HostDatum) {
+ // HostDb entries come before CrawlDb entries
+ return -1;
+ } else {
+ return 1;
+ }
+ }
+ }
+ }
+
+ /**
+ * Mapper that reads HostDb and emits entries for secondary sorting.
+ * Uses a very low score (-Float.MAX_VALUE) to ensure HostDb entries
+ * sort before CrawlDb entries.
+ */
+ public static class HostDbReaderMapper
+ extends Mapper<Text, HostDatum, FloatTextPair, SelectorEntry> {
+
+ @Override
+ public void map(Text hostname, HostDatum value, Context context)
+ throws IOException, InterruptedException {
+ SelectorEntry hostDataSelector = new SelectorEntry();
+ hostDataSelector.hasHostDatum.set(true);
+ try {
+ hostDataSelector.hostdatum = (HostDatum) value.clone();
+ } catch (CloneNotSupportedException e) {
+ hostDataSelector.hostdatum = value;
+ }
+
+ // Use very low score and hostname to ensure HostDb entries
+ // sort before CrawlDb entries for the same host
+ context.write(new FloatTextPair(new FloatWritable(-Float.MAX_VALUE),
+ hostname), hostDataSelector);
+ }
+ }
+
+ /** Selects entries due for fetch. Partitions by host/domain/IP (with HostDb
support). */
+ public static class SelectorWithHostDb extends Partitioner<FloatTextPair,
Writable>
+ implements Configurable {
+
+ private final URLPartitioner partitioner = new URLPartitioner();
+ private final Map<String, Integer> partitionByHost = new HashMap<>();
+ private Configuration conf;
+ private int seed;
+ private int cachedReduceTasks = -1;
+ private String mode = URLPartitioner.PARTITION_MODE_HOST;
+
+ /**
+ * Partition by host / domain or IP.
+ * For HostDb entries (key.second is non-empty), partition by the hostname.
+ * For CrawlDb entries (key.second is empty), use the URLPartitioner.
+ *
+ * Note: When partition.url.mode is not "byHost", HostDb entries may not
+ * be correctly grouped with their corresponding CrawlDb entries. For best
+ * results, use "byHost" mode (the default) when using HostDb.
+ */
+ @Override
+ public int getPartition(FloatTextPair key, Writable value,
+ int numReduceTasks) {
+ SelectorEntry entry = (SelectorEntry) value;
+ // For HostDb entries, use the hostname from the key
+ // For CrawlDb entries, use the URL from the entry
+ if (key.second.getLength() > 0) {
+ // HostDb entry - partition by hostname
+ // This works best with "byHost" mode; other modes may cause
+ // HostDb entries to not be grouped with their CrawlDb entries
+ int hashCode = key.second.toString().hashCode();
+ hashCode ^= seed;
+ return (hashCode & Integer.MAX_VALUE) % numReduceTasks;
+ }
+ if (cachedReduceTasks != numReduceTasks) {
+ partitionByHost.clear();
+ cachedReduceTasks = numReduceTasks;
+ }
+ String host = rawHost(entry.url);
+ if (host != null) {
+ Integer cached = partitionByHost.get(host);
+ if (cached != null) {
+ return cached;
+ }
+ }
+ int part = partitioner.getPartition(entry.url, key.first,
numReduceTasks);
Review Comment:
Agreed, and the cache is gone. It existed because the 50M HostDb profile
showed `URLNormalizers.normalize` inside `SelectorWithHostDb.getPartition`
(about 4% of CPU and about 22% of allocation). The map stored
`URLPartitioner`'s int so later URLs on the same host skipped that work.
Skipping normalization on this path removes the work, so the map is
unnecessary. The no-HostDb selector is unchanged and still calls
`URLPartitioner` for every URL.
> Use secondary sorting for memory-efficient HostDb integration in Generator
> --------------------------------------------------------------------------
>
> Key: NUTCH-2455
> URL: https://issues.apache.org/jira/browse/NUTCH-2455
> Project: Nutch
> Issue Type: Improvement
> Components: generator
> Affects Versions: 1.13
> Reporter: Markus Jelsma
> Assignee: Lewis John McGibbney
> Priority: Major
> Fix For: 1.24
>
> Attachments: NUTCH-2455.patch
>
>
> Citing Sebastian at NUTCH-2420:
> ??The correct solution would be to use <host,score> pairs as keys in the
> Selector job, with a partitioner and secondary sorting so that all keys with
> same host end up in the same call of the reducer. If values can also hold a
> HostDb entry and the sort comparator guarantees that the HostDb entry
> (entries if partitioned by domain or IP) comes in front of all CrawlDb
> entries. But that would be a substantial improvement...??
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
