Index: src/java/org/apache/nutch/fetcher/FetchItemQueue.java =================================================================== --- src/java/org/apache/nutch/fetcher/FetchItemQueue.java (revision 5116) +++ src/java/org/apache/nutch/fetcher/FetchItemQueue.java (working copy) @@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.io.Text; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -46,6 +47,7 @@ long minCrawlDelay; int maxThreads; Configuration conf; + Text cookie; public FetchItemQueue(Configuration conf, int maxThreads, long crawlDelay, long minCrawlDelay) { @@ -85,6 +87,12 @@ public void addFetchItem(FetchItem it) { if (it == null) return; + + // Check for variable crawl delay + if (it.datum.getMetaData().containsKey("_variableFetchDelay_")) { + crawlDelay = it.datum.getMetaData().get("_variableFetchDelay_").get(); + setEndTime(System.currentTimeMillis() - crawlDelay); + } queue.add(it); } @@ -113,6 +121,14 @@ } return it; } + + public void setCookie(Text cookie) { + this.cookie = cookie; + } + + public Text getCookie() { + return cookie; + } public synchronized void dump() { LOG.info(" maxThreads = " + maxThreads); Index: src/java/org/apache/nutch/crawl/Generator.java =================================================================== --- src/java/org/apache/nutch/crawl/Generator.java (revision 5116) +++ src/java/org/apache/nutch/crawl/Generator.java (working copy) @@ -27,6 +27,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.commons.jexl2.Expression; +import org.apache.commons.jexl2.JexlContext; +import org.apache.commons.jexl2.MapContext; import org.apache.hadoop.io.*; import org.apache.hadoop.conf.*; import org.apache.hadoop.mapred.*; @@ -35,6 +37,7 @@ import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; +import org.apache.nutch.hostdb.HostDatum; import org.apache.nutch.metadata.Nutch; import org.apache.nutch.net.URLFilterException; import org.apache.nutch.net.URLFilters; @@ -49,7 +52,6 @@ import org.apache.nutch.util.TimingUtil; import org.apache.nutch.util.URLUtil; - /** * Generates a subset of a crawl db to fetch. This version allows to generate * fetchlists for several segments in one go. Unlike in the initial version @@ -78,6 +80,9 @@ public static final String GENERATOR_DELAY = "crawl.gen.delay"; public static final String GENERATOR_MAX_NUM_SEGMENTS = "generate.max.num.segments"; public static final String GENERATOR_EXPR = "generate.expr"; + public static final String GENERATOR_HOSTDB = "generate.hostdb"; + public static final String GENERATOR_MAX_COUNT_EXPR = "generate.max.count.expr"; + public static final String GENERATOR_FETCH_DELAY_EXPR = "generate.fetch.delay.expr"; public static class SelectorEntry implements Writable { public Text url; @@ -137,8 +142,13 @@ private int maxNumSegments = 1; private Expression expr = null; private int currentsegmentnum = 1; - + private SequenceFile.Reader[] hostdbReaders = null; + private Expression maxCountExpr = null; + private Expression fetchDelayExpr = null; + private JobConf conf = null; + public void configure(JobConf job) { + this.conf = job; curTime = job.getLong(GENERATOR_CUR_TIME, System.currentTimeMillis()); limit = job.getLong(GENERATOR_TOP_N, Long.MAX_VALUE) / job.getNumReduceTasks(); @@ -167,9 +177,25 @@ expr = JexlUtil.parseExpression(job.get(GENERATOR_EXPR, null)); maxNumSegments = job.getInt(GENERATOR_MAX_NUM_SEGMENTS, 1); segCounts = new int[maxNumSegments]; + + if (job.get(GENERATOR_HOSTDB) != null) { + try { + Path path = new Path(job.get(GENERATOR_HOSTDB), "current"); + hostdbReaders = SequenceFileOutputFormat.getReaders(job, path); + maxCountExpr = JexlUtil.parseExpression(job.get(GENERATOR_MAX_COUNT_EXPR, null)); + fetchDelayExpr = JexlUtil.parseExpression(job.get(GENERATOR_FETCH_DELAY_EXPR, null)); + } catch (IOException e) {} + } } public void close() { + if (hostdbReaders != null) { + try { + for (int i = 0; i < hostdbReaders.length; i++) { + hostdbReaders[i].close(); + } + } catch (IOException e) {} + } } /** Select & invert subset due for fetch. */ @@ -252,13 +278,101 @@ return partitioner.getPartition(((SelectorEntry) value).url, key, numReduceTasks); } + + private HostDatum getHostDatum(String host) throws Exception { + Text key = new Text(); + HostDatum value = new HostDatum(); + + for (int i = 0; i < hostdbReaders.length; i++) { + while (hostdbReaders[i].next(key, value)) { + if (host.equals(key.toString())) { + return value; + } + } + } + + return null; + } + + private JexlContext createContext(HostDatum datum) { + JexlContext context = new MapContext(); + context.set("dnsFailures", datum.getDnsFailures()); + context.set("connectionFailures", datum.getConnectionFailures()); + context.set("unfetched", datum.getUnfetched()); + context.set("notModified", datum.getNotModified()); + context.set("redirTemp", datum.getRedirTemp()); + context.set("redirPerm", datum.getRedirPerm()); + context.set("gone", datum.getGone()); + context.set("conf", conf); + + // Set metadata variables + for (Map.Entry entry : datum.getMetaData().entrySet()) { + Object value = entry.getValue(); + + if (value instanceof FloatWritable) { + FloatWritable fvalue = (FloatWritable)value; + Text tkey = (Text)entry.getKey(); + context.set(tkey.toString(), fvalue.get()); + } + + if (value instanceof IntWritable) { + IntWritable ivalue = (IntWritable)value; + Text tkey = (Text)entry.getKey(); + context.set(tkey.toString(), ivalue.get()); + } + + if (value instanceof Text) { + Text tvalue = (Text)value; + Text tkey = (Text)entry.getKey(); + context.set(tkey.toString().replace("-", "_"), tvalue.toString()); + } + } + + return context; + } /** Collect until limit is reached. */ public void reduce(FloatWritable key, Iterator values, OutputCollector output, Reporter reporter) throws IOException { - + + String hostname = null; + HostDatum host = null; + long variableMaxCount = 0l; + long variableFetchDelay = 0l; // in millis + while (values.hasNext()) { + SelectorEntry entry = values.next(); + Text url = entry.url; + String urlString = url.toString(); + URL u = null; + + // Do this only once per queue + if (host == null) { + try { + hostname = URLUtil.getHost(urlString); + host = getHostDatum(hostname); + } catch (Exception e) {} + + // Got it? + if (host == null) { + // Didn't work, prevent future lookups + host = new HostDatum(); + } else { + if (maxCountExpr != null) { + variableMaxCount = Math.round((double)maxCountExpr.evaluate(createContext(host))); + LOG.info("Generator: variable maxCount: " + variableMaxCount + " for " + hostname); + maxCount = (int)variableMaxCount; + } + + if (fetchDelayExpr != null) { + variableFetchDelay = Math.round((double)fetchDelayExpr.evaluate(createContext(host))); + LOG.info("Generator: variable fetchDelay: " + variableFetchDelay + " for " + hostname); + entry.datum.getMetaData().put(new Text("_variableFetchDelay_"), new LongWritable(variableFetchDelay)); + System.out.println(entry.datum); + } + } + } if (count == limit) { // do we have any segments left? @@ -269,11 +383,6 @@ break; } - SelectorEntry entry = values.next(); - Text url = entry.url; - String urlString = url.toString(); - URL u = null; - String hostordomain = null; try { @@ -503,6 +612,13 @@ return generate(dbDir, segments, numLists, topN, curTime, filter, true, force, 1, null); } + + public Path[] generate(Path dbDir, Path segments, int numLists, long topN, + long curTime, boolean filter, boolean norm, boolean force, + int maxNumSegments, String expr) throws IOException { + return generate(dbDir, segments, numLists, topN, curTime, filter, true, + force, 1, expr, null); + } /** * Generate fetchlists in one or more segments. Whether to filter URLs or not @@ -528,7 +644,7 @@ */ public Path[] generate(Path dbDir, Path segments, int numLists, long topN, long curTime, boolean filter, boolean norm, boolean force, - int maxNumSegments, String expr) throws IOException { + int maxNumSegments, String expr, String hostdb) throws IOException { Path tempDir = new Path(getConf().get("mapred.temp.dir", ".") + "/generate-temp-" + java.util.UUID.randomUUID().toString()); @@ -549,6 +665,9 @@ if (expr != null) { LOG.info("Generator: expr: " + expr); } + if (expr != null) { + LOG.info("Generator: hostdb: " + hostdb); + } // map to inverted subset due for fetch, sort by score JobConf job = new NutchJob(getConf()); @@ -573,6 +692,9 @@ if (expr != null) { job.set(GENERATOR_EXPR, expr); } + if (hostdb != null) { + job.set(GENERATOR_HOSTDB, hostdb); + } FileInputFormat.addInputPath(job, new Path(dbDir, CrawlDb.CURRENT_NAME)); job.setInputFormat(SequenceFileInputFormat.class); @@ -729,6 +851,7 @@ Path dbDir = new Path(args[0]); Path segmentsDir = new Path(args[1]); + String hostdb = null; long curTime = System.currentTimeMillis(); long topN = Long.MAX_VALUE; int numFetchers = -1; @@ -745,6 +868,9 @@ } else if ("-numFetchers".equals(args[i])) { numFetchers = Integer.parseInt(args[i + 1]); i++; + } else if ("-hostdb".equals(args[i])) { + hostdb = args[i + 1]; + i++; } else if ("-adddays".equals(args[i])) { long numDays = Integer.parseInt(args[i + 1]); curTime += numDays * 1000L * 60 * 60 * 24; @@ -764,7 +890,7 @@ try { Path[] segs = generate(dbDir, segmentsDir, numFetchers, topN, curTime, - filter, norm, force, maxNumSegments, expr); + filter, norm, force, maxNumSegments, expr, hostdb); if (segs == null) return 1; } catch (Exception e) {