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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
