Author: ab
Date: Wed Aug 18 12:34:36 2010
New Revision: 986647
URL: http://svn.apache.org/viewvc?rev=986647&view=rev
Log:
NUTCH-884 FetcherJob should run more reduce tasks than default.
Modified:
nutch/trunk/CHANGES.txt
nutch/trunk/src/java/org/apache/nutch/fetcher/FetcherJob.java
nutch/trunk/src/java/org/apache/nutch/tools/Benchmark.java
nutch/trunk/src/test/org/apache/nutch/fetcher/TestFetcher.java
Modified: nutch/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/nutch/trunk/CHANGES.txt?rev=986647&r1=986646&r2=986647&view=diff
==============================================================================
--- nutch/trunk/CHANGES.txt (original)
+++ nutch/trunk/CHANGES.txt Wed Aug 18 12:34:36 2010
@@ -2,6 +2,8 @@ Nutch Change Log
Release 2.0 - Current Development
+* NUTCH-884 FetcherJob should run more reduce tasks than default (ab)
+
* NUTCH-883 Remove unused parameters from nutch-default.xml (jnioche)
* NUTCH-886 A .gitignore file for Nutch (dogacan)
Modified: nutch/trunk/src/java/org/apache/nutch/fetcher/FetcherJob.java
URL:
http://svn.apache.org/viewvc/nutch/trunk/src/java/org/apache/nutch/fetcher/FetcherJob.java?rev=986647&r1=986646&r2=986647&view=diff
==============================================================================
--- nutch/trunk/src/java/org/apache/nutch/fetcher/FetcherJob.java (original)
+++ nutch/trunk/src/java/org/apache/nutch/fetcher/FetcherJob.java Wed Aug 18
12:34:36 2010
@@ -140,15 +140,17 @@ public class FetcherJob implements Tool
/**
* Run fetcher.
- * @param threads number of threads per map task
* @param crawlId crawlId (obtained from Generator) or null to fetch all
generated fetchlists
+ * @param threads number of threads per map task
* @param shouldResume
* @param parse if true, then parse content immediately, if false then a
separate
* run of {...@link ParserJob} will be needed.
+ * @param numTasks number of fetching tasks (reducers). If set to < 1 then
use the default,
+ * which is mapred.map.tasks.
* @return 0 on success
* @throws Exception
*/
- public int fetch(int threads, String crawlId, boolean shouldResume, boolean
parse)
+ public int fetch(String crawlId, int threads, boolean shouldResume, boolean
parse, int numTasks)
throws Exception {
LOG.info("FetcherJob: starting");
@@ -181,11 +183,16 @@ public class FetcherJob implements Tool
}
Job job = new NutchJob(getConf(), "fetch");
-
Collection<WebPage.Field> fields = getFields(job);
StorageUtils.initMapperJob(job, fields, IntWritable.class,
FetchEntry.class, FetcherMapper.class, PartitionUrlByHost.class,
false);
StorageUtils.initReducerJob(job, FetcherReducer.class);
+ if (numTasks < 1) {
+ job.setNumReduceTasks(job.getConfiguration().getInt("mapred.map.tasks",
+ job.getNumReduceTasks()));
+ } else {
+ job.setNumReduceTasks(numTasks);
+ }
boolean success = job.waitForCompletion(true);
if (!success) {
@@ -237,7 +244,12 @@ public class FetcherJob implements Tool
boolean parse = getConf().getBoolean(PARSE_KEY, false);
String crawlId;
- String usage = "Usage: FetcherJob (<crawl id> | -all) [-threads N]
[-parse] [-resume]";
+ String usage = "Usage: FetcherJob (<crawl id> | -all) [-threads N]
[-parse] [-resume] [-numTasks N]\n" +
+ "\tcrawlId\tcrawl identifier returned by Generator, or -all for all
generated crawlId-s\n" +
+ "\t-threads N\tnumber of fetching threads per task\n" +
+ "\t-parse\tif specified then fetcher will immediately parse fetched
content\n" +
+ "\t-resume\tresume interrupted job\n" +
+ "\t-numTasks N\tif N > 0 then use this many reduce tasks for fetching
(default: mapred.map.tasks)";
if (args.length == 0) {
System.err.println(usage);
@@ -245,10 +257,11 @@ public class FetcherJob implements Tool
}
crawlId = args[0];
- if (crawlId.equals("-threads") || crawlId.equals("-resume") ||
crawlId.equals("-parse")) {
+ if (!crawlId.equals("-all") || crawlId.startsWith("-")) {
System.err.println(usage);
return -1;
}
+ int numTasks = -1;
for (int i = 1; i < args.length; i++) {
if ("-threads".equals(args[i])) {
// found -threads option
@@ -257,10 +270,12 @@ public class FetcherJob implements Tool
shouldResume = true;
} else if ("-parse".equals(args[i])) {
parse = true;
+ } else if ("-numTasks".equals(args[i])) {
+ numTasks = Integer.parseInt(args[++i]);
}
}
- int fetchcode = fetch(threads, crawlId, shouldResume, parse); // run the
Fetcher
+ int fetchcode = fetch(crawlId, threads, shouldResume, parse, numTasks); //
run the Fetcher
return fetchcode;
}
Modified: nutch/trunk/src/java/org/apache/nutch/tools/Benchmark.java
URL:
http://svn.apache.org/viewvc/nutch/trunk/src/java/org/apache/nutch/tools/Benchmark.java?rev=986647&r1=986646&r2=986647&view=diff
==============================================================================
--- nutch/trunk/src/java/org/apache/nutch/tools/Benchmark.java (original)
+++ nutch/trunk/src/java/org/apache/nutch/tools/Benchmark.java Wed Aug 18
12:34:36 2010
@@ -203,7 +203,7 @@ public class Benchmark extends Configure
}
boolean isParsing = getConf().getBoolean("fetcher.parse", true);
start = System.currentTimeMillis();
- fetcher.fetch(threads, crawlId, false, isParsing); // fetch it
+ fetcher.fetch(crawlId, threads, false, isParsing, -1); // fetch it
delta = System.currentTimeMillis() - start;
res.addTiming("fetch", i + "", delta);
if (!isParsing) {
Modified: nutch/trunk/src/test/org/apache/nutch/fetcher/TestFetcher.java
URL:
http://svn.apache.org/viewvc/nutch/trunk/src/test/org/apache/nutch/fetcher/TestFetcher.java?rev=986647&r1=986646&r2=986647&view=diff
==============================================================================
--- nutch/trunk/src/test/org/apache/nutch/fetcher/TestFetcher.java (original)
+++ nutch/trunk/src/test/org/apache/nutch/fetcher/TestFetcher.java Wed Aug 18
12:34:36 2010
@@ -85,7 +85,7 @@ public class TestFetcher extends Abstrac
time = System.currentTimeMillis();
conf.setBoolean(FetcherJob.PARSE_KEY, true);
FetcherJob fetcher = new FetcherJob(conf);
- fetcher.fetch(1, crawlId, false, true);
+ fetcher.fetch(crawlId, 1, false, true, -1);
time = System.currentTimeMillis() - time;
