sebastian-nagel commented on code in PR #888:
URL: https://github.com/apache/nutch/pull/888#discussion_r4431106493
##########
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:
Looks like `URLUtil.getUrlRootByMode` is used directly everywhere.
##########
src/java/org/apache/nutch/crawl/URLPartitioner.java:
##########
Review Comment:
Could also simplify by calling `URLUtil.getUrlRootByMode` and call
`hashCode()` on the result.
##########
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:
(could be simplified if `byHost` is guaranteed and no normalization is done)
The host name is already extracted: create a method `getPartionForRoot` and
call it directly.
##########
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:
Could also make this either an error or just set the mode to `byHost`.
If `byHost` is guaranteed, the code could be simplified and improved for
better performance.
##########
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:
This duplicates the logic already present in URLPartitioner. Could move it
to a function/method and call it from here.
##########
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:
Is this method used at all?
##########
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:
Even the cache `partitionByHost` would be obsolete then (if `byHost` and no
normalization).
--
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]
