You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nutch.apache.org by fe...@apache.org on 2012/03/05 13:57:05 UTC
svn commit: r1297039 - in /nutch/branches/nutchgora: ./
src/java/org/apache/nutch/crawl/ src/java/org/apache/nutch/fetcher/
src/test/org/apache/nutch/crawl/
Author: ferdy
Date: Mon Mar 5 12:57:05 2012
New Revision: 1297039
URL: http://svn.apache.org/viewvc?rev=1297039&view=rev
Log:
NUTCH-1289 In distributed mode URL's are not partitioned
Added:
nutch/branches/nutchgora/src/test/org/apache/nutch/crawl/TestURLPartitioner.java (with props)
Removed:
nutch/branches/nutchgora/src/java/org/apache/nutch/fetcher/PartitionUrlByHost.java
Modified:
nutch/branches/nutchgora/CHANGES.txt
nutch/branches/nutchgora/src/java/org/apache/nutch/crawl/GeneratorJob.java
nutch/branches/nutchgora/src/java/org/apache/nutch/crawl/URLPartitioner.java
nutch/branches/nutchgora/src/java/org/apache/nutch/fetcher/FetcherJob.java
Modified: nutch/branches/nutchgora/CHANGES.txt
URL: http://svn.apache.org/viewvc/nutch/branches/nutchgora/CHANGES.txt?rev=1297039&r1=1297038&r2=1297039&view=diff
==============================================================================
--- nutch/branches/nutchgora/CHANGES.txt (original)
+++ nutch/branches/nutchgora/CHANGES.txt Mon Mar 5 12:57:05 2012
@@ -2,6 +2,8 @@ Nutch Change Log
Release nutchgora - Current Development
+* NUTCH-1289 In distributed mode URL's are not partitioned (Dan Rosher, ferdy)
+
* NUTCH-1292 Better exception logging and debugging during fetch. (ferdy)
* NUTCH-1263 FetcherJob must put 'fetchTime' on input (ferdy)
Modified: nutch/branches/nutchgora/src/java/org/apache/nutch/crawl/GeneratorJob.java
URL: http://svn.apache.org/viewvc/nutch/branches/nutchgora/src/java/org/apache/nutch/crawl/GeneratorJob.java?rev=1297039&r1=1297038&r2=1297039&view=diff
==============================================================================
--- nutch/branches/nutchgora/src/java/org/apache/nutch/crawl/GeneratorJob.java (original)
+++ nutch/branches/nutchgora/src/java/org/apache/nutch/crawl/GeneratorJob.java Mon Mar 5 12:57:05 2012
@@ -19,8 +19,6 @@ package org.apache.nutch.crawl;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
-import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Random;
@@ -29,14 +27,13 @@ import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
-import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
+import org.apache.nutch.crawl.URLPartitioner.SelectorEntryPartitioner;
import org.apache.nutch.metadata.Nutch;
import org.apache.nutch.storage.StorageUtils;
import org.apache.nutch.storage.WebPage;
@@ -178,7 +175,7 @@ public class GeneratorJob extends NutchT
currentJobNum = 0;
currentJob = new NutchJob(getConf(), "generate: " + batchId);
StorageUtils.initMapperJob(currentJob, FIELDS, SelectorEntry.class,
- WebPage.class, GeneratorMapper.class, URLPartitioner.class, true);
+ WebPage.class, GeneratorMapper.class, SelectorEntryPartitioner.class, true);
StorageUtils.initReducerJob(currentJob, GeneratorReducer.class);
currentJob.waitForCompletion(true);
ToolUtil.recordJobStatus(null, currentJob, results);
Modified: nutch/branches/nutchgora/src/java/org/apache/nutch/crawl/URLPartitioner.java
URL: http://svn.apache.org/viewvc/nutch/branches/nutchgora/src/java/org/apache/nutch/crawl/URLPartitioner.java?rev=1297039&r1=1297038&r2=1297039&view=diff
==============================================================================
--- nutch/branches/nutchgora/src/java/org/apache/nutch/crawl/URLPartitioner.java (original)
+++ nutch/branches/nutchgora/src/java/org/apache/nutch/crawl/URLPartitioner.java Mon Mar 5 12:57:05 2012
@@ -26,19 +26,20 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.nutch.crawl.GeneratorJob.SelectorEntry;
+import org.apache.nutch.fetcher.FetchEntry;
import org.apache.nutch.net.URLNormalizers;
import org.apache.nutch.storage.WebPage;
+import org.apache.nutch.util.TableUtil;
import org.apache.nutch.util.URLUtil;
/**
* Partition urls by host, domain name or IP depending on the value of the
* parameter 'partition.url.mode' which can be 'byHost', 'byDomain' or 'byIP'
*/
-public class URLPartitioner
-extends Partitioner<SelectorEntry, WebPage>
-implements Configurable {
+public class URLPartitioner implements Configurable {
private static final Logger LOG = LoggerFactory.getLogger(URLPartitioner.class);
public static final String PARTITION_MODE_KEY = "partition.url.mode";
@@ -46,6 +47,8 @@ implements Configurable {
public static final String PARTITION_MODE_HOST = "byHost";
public static final String PARTITION_MODE_DOMAIN = "byDomain";
public static final String PARTITION_MODE_IP = "byIP";
+
+ public static final String PARTITION_URL_SEED = "partition.url.seed";
private Configuration conf;
@@ -61,7 +64,7 @@ implements Configurable {
@Override
public void setConf(Configuration conf) {
this.conf = conf;
- seed = conf.getInt("partition.url.seed", 0);
+ seed = conf.getInt(PARTITION_URL_SEED, 0);
mode = conf.get(PARTITION_MODE_KEY, PARTITION_MODE_HOST);
// check that the mode is known
if (!mode.equals(PARTITION_MODE_IP) && !mode.equals(PARTITION_MODE_DOMAIN)
@@ -72,37 +75,88 @@ implements Configurable {
normalizers = new URLNormalizers(conf, URLNormalizers.SCOPE_PARTITION);
}
- public void setup(Configuration conf) {
-
- }
-
- @Override
- public int getPartition(SelectorEntry key, WebPage value, int numReduceTasks) {
- String urlString = key.url;
+ public int getPartition(String urlString, int numReduceTasks) {
+ if (numReduceTasks == 1) {
+ //this check can be removed when we use Hadoop with MAPREDUCE-1287
+ return 0;
+ }
+
+ int hashCode;
URL url = null;
- int hashCode = urlString.hashCode();
try {
urlString = normalizers.normalize(urlString, URLNormalizers.SCOPE_PARTITION);
+ hashCode = urlString.hashCode();
url = new URL(urlString);
- hashCode = url.getHost().hashCode();
} catch (MalformedURLException e) {
LOG.warn("Malformed URL: '" + urlString + "'");
+ hashCode = urlString.hashCode();
}
-
- if (mode.equals(PARTITION_MODE_DOMAIN) && url != null) hashCode = URLUtil
- .getDomainName(url).hashCode();
- else if (mode.equals(PARTITION_MODE_IP)) {
- try {
- InetAddress address = InetAddress.getByName(url.getHost());
- hashCode = address.getHostAddress().hashCode();
- } catch (UnknownHostException e) {
- GeneratorJob.LOG.info("Couldn't find IP for host: " + url.getHost());
+
+ if (url != null) {
+ if (mode.equals(PARTITION_MODE_HOST)) {
+ hashCode = url.getHost().hashCode();
+ } else if (mode.equals(PARTITION_MODE_DOMAIN)) {
+ hashCode = URLUtil.getDomainName(url).hashCode();
+ } else { // MODE IP
+ try {
+ InetAddress address = InetAddress.getByName(url.getHost());
+ hashCode = address.getHostAddress().hashCode();
+ } catch (UnknownHostException e) {
+ GeneratorJob.LOG.info("Couldn't find IP for host: " + url.getHost());
+ }
}
}
-
+
// make hosts wind up in different partitions on different runs
hashCode ^= seed;
-
return (hashCode & Integer.MAX_VALUE) % numReduceTasks;
}
+
+
+ public static class SelectorEntryPartitioner
+ extends Partitioner<SelectorEntry, WebPage> implements Configurable {
+ private URLPartitioner partitioner = new URLPartitioner();
+ private Configuration conf;
+
+ @Override
+ public int getPartition(SelectorEntry selectorEntry, WebPage page, int numReduces) {
+ return partitioner.getPartition(selectorEntry.url, numReduces);
+ }
+
+ @Override
+ public Configuration getConf() {
+ return conf;
+ }
+
+ @Override
+ public void setConf(Configuration conf) {
+ this.conf=conf;
+ partitioner.setConf(conf);
+ }
+ }
+
+ public static class FetchEntryPartitioner
+ extends Partitioner<IntWritable, FetchEntry> implements Configurable {
+ private URLPartitioner partitioner = new URLPartitioner();
+ private Configuration conf;
+
+ @Override
+ public int getPartition(IntWritable intWritable, FetchEntry fetchEntry, int numReduces) {
+ String key = fetchEntry.getKey();
+ String url = TableUtil.unreverseUrl(key);
+ return partitioner.getPartition(url, numReduces);
+ }
+
+ @Override
+ public Configuration getConf() {
+ return conf;
+ }
+
+ @Override
+ public void setConf(Configuration conf) {
+ this.conf=conf;
+ partitioner.setConf(conf);
+ }
+ }
+
}
Modified: nutch/branches/nutchgora/src/java/org/apache/nutch/fetcher/FetcherJob.java
URL: http://svn.apache.org/viewvc/nutch/branches/nutchgora/src/java/org/apache/nutch/fetcher/FetcherJob.java?rev=1297039&r1=1297038&r2=1297039&view=diff
==============================================================================
--- nutch/branches/nutchgora/src/java/org/apache/nutch/fetcher/FetcherJob.java (original)
+++ nutch/branches/nutchgora/src/java/org/apache/nutch/fetcher/FetcherJob.java Mon Mar 5 12:57:05 2012
@@ -19,8 +19,6 @@ package org.apache.nutch.fetcher;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Random;
@@ -35,6 +33,7 @@ import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.nutch.crawl.GeneratorJob;
+import org.apache.nutch.crawl.URLPartitioner.FetchEntryPartitioner;
import org.apache.nutch.metadata.Nutch;
import org.apache.nutch.parse.ParserJob;
import org.apache.nutch.protocol.ProtocolFactory;
@@ -184,7 +183,7 @@ public class FetcherJob extends NutchToo
currentJob = new NutchJob(getConf(), "fetch");
Collection<WebPage.Field> fields = getFields(currentJob);
StorageUtils.initMapperJob(currentJob, fields, IntWritable.class,
- FetchEntry.class, FetcherMapper.class, PartitionUrlByHost.class, false);
+ FetchEntry.class, FetcherMapper.class, FetchEntryPartitioner.class, false);
StorageUtils.initReducerJob(currentJob, FetcherReducer.class);
if (numTasks == null || numTasks < 1) {
currentJob.setNumReduceTasks(currentJob.getConfiguration().getInt("mapred.map.tasks",
Added: nutch/branches/nutchgora/src/test/org/apache/nutch/crawl/TestURLPartitioner.java
URL: http://svn.apache.org/viewvc/nutch/branches/nutchgora/src/test/org/apache/nutch/crawl/TestURLPartitioner.java?rev=1297039&view=auto
==============================================================================
--- nutch/branches/nutchgora/src/test/org/apache/nutch/crawl/TestURLPartitioner.java (added)
+++ nutch/branches/nutchgora/src/test/org/apache/nutch/crawl/TestURLPartitioner.java Mon Mar 5 12:57:05 2012
@@ -0,0 +1,205 @@
+/*******************************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ ******************************************************************************/
+package org.apache.nutch.crawl;
+
+import java.net.MalformedURLException;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.nutch.crawl.GeneratorJob.SelectorEntry;
+import org.apache.nutch.crawl.URLPartitioner.FetchEntryPartitioner;
+import org.apache.nutch.crawl.URLPartitioner.SelectorEntryPartitioner;
+import org.apache.nutch.fetcher.FetchEntry;
+import org.apache.nutch.storage.WebPage;
+import org.apache.nutch.util.NutchConfiguration;
+import org.apache.nutch.util.TableUtil;
+
+/**
+ * Tests {@link URLPartitioner}
+ */
+public class TestURLPartitioner extends TestCase {
+
+ /**
+ * tests one reducer, everything goes into one partition, using host partitioner.
+ */
+ public void testOneReducer() {
+ URLPartitioner partitioner = new URLPartitioner();
+ Configuration conf = NutchConfiguration.create();
+ conf.set(URLPartitioner.PARTITION_MODE_KEY, URLPartitioner.PARTITION_MODE_HOST);
+ partitioner.setConf(conf);
+
+ int numReduceTasks = 1;
+
+ assertEquals(0, partitioner.getPartition("http://example.org", numReduceTasks));
+ assertEquals(0, partitioner.getPartition("http://www.apache.org", numReduceTasks));
+ }
+
+ /**
+ * tests partitioning by host
+ */
+ public void testModeHost() {
+ URLPartitioner partitioner = new URLPartitioner();
+ Configuration conf = NutchConfiguration.create();
+ conf.set(URLPartitioner.PARTITION_MODE_KEY, URLPartitioner.PARTITION_MODE_HOST);
+ partitioner.setConf(conf);
+
+ int numReduceTasks = 100;
+
+ int partitionWithoutWWW = partitioner.getPartition("http://example.org/", numReduceTasks);
+ int partitionWithWWW = partitioner.getPartition("http://www.example.org/", numReduceTasks);
+ assertNotSame("partitions should differ because of different host",
+ partitionWithoutWWW, partitionWithWWW);
+
+ int partitionSame1 = partitioner.getPartition("http://www.example.org/paris", numReduceTasks);
+ int partitionSame2 = partitioner.getPartition("http://www.example.org/london", numReduceTasks);
+ assertEquals("partitions should be same because of same host",
+ partitionSame1, partitionSame2);
+ }
+
+ /**
+ * tests partitioning by domain
+ */
+ public void testModeDomain() {
+ URLPartitioner partitioner = new URLPartitioner();
+ Configuration conf = NutchConfiguration.create();
+ conf.set(URLPartitioner.PARTITION_MODE_KEY, URLPartitioner.PARTITION_MODE_DOMAIN);
+ partitioner.setConf(conf);
+
+ int numReduceTasks = 100;
+
+ int partitionExample = partitioner.getPartition("http://www.example.org/", numReduceTasks);
+ int partitionApache = partitioner.getPartition("http://www.apache.org/", numReduceTasks);
+ assertNotSame("partitions should differ because of different domain",
+ partitionExample, partitionApache);
+
+ int partitionWithoutWWW = partitioner.getPartition("http://example.org/", numReduceTasks);
+ int partitionWithWWW = partitioner.getPartition("http://www.example.org/", numReduceTasks);
+ assertEquals("partitions should be same because of same domain",
+ partitionWithoutWWW, partitionWithWWW);
+ }
+
+ /**
+ * tests partitioning by IP
+ */
+ public void testModeIP() {
+ URLPartitioner partitioner = new URLPartitioner();
+ Configuration conf = NutchConfiguration.create();
+ conf.set(URLPartitioner.PARTITION_MODE_KEY, URLPartitioner.PARTITION_MODE_IP);
+ partitioner.setConf(conf);
+
+ int numReduceTasks = 100;
+
+ int partitionExample = partitioner.getPartition("http://www.example.org/", numReduceTasks);
+ int partitionApache = partitioner.getPartition("http://www.apache.org/", numReduceTasks);
+ assertNotSame("partitions should differ because of different ip",
+ partitionExample, partitionApache);
+
+ int partitionWithoutWWW = partitioner.getPartition("http://example.org/", numReduceTasks);
+ int partitionWithWWW = partitioner.getPartition("http://www.example.org/", numReduceTasks);
+ //the following has dependendy on example.org (that is has the same ip as www.example.org)
+ assertEquals("partitions should be same because of same ip",
+ partitionWithoutWWW, partitionWithWWW);
+ }
+
+
+ /**
+ * Test the seed functionality, using host partitioner.
+ */
+ public void testSeed() {
+ URLPartitioner partitioner = new URLPartitioner();
+ Configuration conf = NutchConfiguration.create();
+ conf.set(URLPartitioner.PARTITION_MODE_KEY, URLPartitioner.PARTITION_MODE_HOST);
+ partitioner.setConf(conf);
+
+ int numReduceTasks = 100;
+ int partitionNoSeed = partitioner.getPartition("http://example.org/", numReduceTasks);
+
+ conf.setInt(URLPartitioner.PARTITION_URL_SEED, 1);
+ partitioner.setConf(conf);
+
+ int partitionWithSeed = partitioner.getPartition("http://example.org/", numReduceTasks);
+
+ assertNotSame("partitions should differ because of different seed",
+ partitionNoSeed, partitionWithSeed);
+ }
+
+
+ /**
+ * Tests the {@link SelectorEntryPartitioner}.
+ */
+ public void testSelectorEntryPartitioner() {
+ //The reference partitioner
+ URLPartitioner refPartitioner = new URLPartitioner();
+
+ //The to be tested partitioner with specific signature
+ URLPartitioner.SelectorEntryPartitioner sigPartitioner =
+ new URLPartitioner.SelectorEntryPartitioner();
+
+ Configuration conf = NutchConfiguration.create();
+ conf.set(URLPartitioner.PARTITION_MODE_KEY, URLPartitioner.PARTITION_MODE_HOST);
+
+ refPartitioner.setConf(conf);
+ sigPartitioner.setConf(conf);
+
+ int numReduceTasks = 100;
+
+ int partitionFromRef = refPartitioner.getPartition("http://www.example.org/", numReduceTasks);
+ //init selector entry (score shouldn't matter)
+ SelectorEntry selectorEntry = new SelectorEntry("http://www.example.org/", 1337);
+ WebPage page = new WebPage();
+ int partitionFromSig = sigPartitioner.getPartition(selectorEntry, page, numReduceTasks);
+
+ assertEquals("partitions should be same",
+ partitionFromRef, partitionFromSig);
+
+ }
+
+ /**
+ * Tests the {@link FetchEntryPartitioner}
+ * @throws MalformedURLException
+ */
+ public void testFetchEntryPartitioner() throws MalformedURLException {
+ //The reference partitioner
+ URLPartitioner refPartitioner = new URLPartitioner();
+
+ //The to be tested partitioner with specific signature
+ URLPartitioner.FetchEntryPartitioner sigPartitioner =
+ new URLPartitioner.FetchEntryPartitioner();
+
+ Configuration conf = NutchConfiguration.create();
+ conf.set(URLPartitioner.PARTITION_MODE_KEY, URLPartitioner.PARTITION_MODE_HOST);
+
+ refPartitioner.setConf(conf);
+ sigPartitioner.setConf(conf);
+
+ int numReduceTasks = 100;
+
+ int partitionFromRef = refPartitioner.getPartition("http://www.example.org/", numReduceTasks);
+ IntWritable intWritable = new IntWritable(1337); //doesn't matter
+ WebPage page = new WebPage();
+ String key = TableUtil.reverseUrl("http://www.example.org/");
+ FetchEntry fetchEntry = new FetchEntry(conf, key, page);
+ int partitionFromSig = sigPartitioner.getPartition(intWritable, fetchEntry, numReduceTasks);
+
+ assertEquals("partitions should be same",
+ partitionFromRef, partitionFromSig);
+
+ }
+
+}
Propchange: nutch/branches/nutchgora/src/test/org/apache/nutch/crawl/TestURLPartitioner.java
------------------------------------------------------------------------------
svn:mime-type = text/plain