You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nutch.apache.org by cu...@apache.org on 2005/05/19 21:53:02 UTC
svn commit: r170983 - in /incubator/nutch/branches/mapred: ./ bin/
src/java/org/apache/nutch/crawl/ src/java/org/apache/nutch/mapred/
src/java/org/apache/nutch/mapred/demo/ src/java/org/apache/nutch/util/
Author: cutting
Date: Thu May 19 12:53:00 2005
New Revision: 170983
URL: http://svn.apache.org/viewcvs?rev=170983&view=rev
Log:
First steps towards a MapReduce-based crawler.
Added:
incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/
incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDBPartitioner.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDBReducer.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDatum.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/DmozParser.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/Injector.java
- copied, changed from r170616, incubator/nutch/branches/mapred/src/java/org/apache/nutch/db/WebDBInjector.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/package.html
incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConfigurable.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConfigured.java
Modified:
incubator/nutch/branches/mapred/bin/nutch
incubator/nutch/branches/mapred/build.xml
incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/InputFormatBase.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobClient.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobConf.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobTracker.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskRunner.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskTracker.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/demo/Grep.java
incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConf.java
Modified: incubator/nutch/branches/mapred/bin/nutch
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/bin/nutch?rev=170983&r1=170982&r2=170983&view=diff
==============================================================================
--- incubator/nutch/branches/mapred/bin/nutch (original)
+++ incubator/nutch/branches/mapred/bin/nutch Thu May 19 12:53:00 2005
@@ -172,9 +172,9 @@
elif [ "$COMMAND" = "ndfs" ] ; then
CLASS=org.apache.nutch.fs.TestClient
elif [ "$COMMAND" = "jobtracker" ] ; then
- CLASS=org.apache.nutch.mapReduce.JobTracker
+ CLASS=org.apache.nutch.mapred.JobTracker
elif [ "$COMMAND" = "tasktracker" ] ; then
- CLASS=org.apache.nutch.mapReduce.TaskTracker
+ CLASS=org.apache.nutch.mapred.TaskTracker
else
CLASS=$COMMAND
fi
Modified: incubator/nutch/branches/mapred/build.xml
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/build.xml?rev=170983&r1=170982&r2=170983&view=diff
==============================================================================
--- incubator/nutch/branches/mapred/build.xml (original)
+++ incubator/nutch/branches/mapred/build.xml Thu May 19 12:53:00 2005
@@ -52,7 +52,7 @@
<!-- ====================================================== -->
<!-- Compile the Java files -->
<!-- ====================================================== -->
- <target name="compile" depends="compile-core, compile-plugins, jar"/>
+ <target name="compile" depends="compile-core, compile-plugins"/>
<target name="compile-core" depends="init">
<javac
@@ -97,12 +97,11 @@
todir="${build.classes}"/>
<copy file="${conf.dir}/nutch-site.xml"
todir="${build.classes}"/>
- <jar
- jarfile="${build.dir}/${final.name}.jar"
- basedir="${build.classes}"
- />
- <symlink overwrite="true" link="${basedir}/${name}.jar"
- resource="${build.dir}/${final.name}.jar"/>
+ <jar jarfile="${build.dir}/${final.name}.jar"
+ basedir="${build.classes}">
+ <manifest>
+ </manifest>
+ </jar>
</target>
<!-- ================================================================== -->
@@ -420,7 +419,6 @@
<!-- ================================================================== -->
<target name="clean">
<delete dir="${build.dir}"/>
- <exec executable="rm"><arg value="-f"/><arg value="${name}.jar"/></exec>
</target>
</project>
Added: incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDBPartitioner.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDBPartitioner.java?rev=170983&view=auto
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDBPartitioner.java (added)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDBPartitioner.java Thu May 19 12:53:00 2005
@@ -0,0 +1,45 @@
+/**
+ * Copyright 2005 The Apache Software Foundation
+ *
+ * Licensed 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.URL;
+import java.net.MalformedURLException;
+
+import org.apache.nutch.io.*;
+import org.apache.nutch.mapred.*;
+
+/** Partition urls by hostname. */
+public class CrawlDBPartitioner implements Partitioner {
+
+ public void configure(JobConf job) {}
+
+ /** Hash by hostname. */
+ public int getPartition(WritableComparable key, int numReduceTasks) {
+ String urlString = ((UTF8)key).toString();
+ URL url = null;
+ try {
+ url = new URL(urlString);
+ } catch (MalformedURLException e) {
+ }
+ int hashCode = (url==null ? urlString : url.getHost()).hashCode();
+
+ return (hashCode & Integer.MAX_VALUE) % numReduceTasks;
+ }
+
+}
+
+
Added: incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDBReducer.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDBReducer.java?rev=170983&view=auto
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDBReducer.java (added)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDBReducer.java Thu May 19 12:53:00 2005
@@ -0,0 +1,42 @@
+/**
+ * Copyright 2005 The Apache Software Foundation
+ *
+ * Licensed 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.URL;
+import java.util.Iterator;
+import java.io.IOException;
+
+import org.apache.nutch.io.*;
+import org.apache.nutch.mapred.*;
+
+/** Merge new page entries with existing entries. */
+public class CrawlDBReducer implements Reducer {
+
+ public void configure(JobConf job) {}
+
+ public void reduce(WritableComparable key, Iterator values,
+ OutputCollector output) throws IOException {
+ // collect datum with the highest status
+ CrawlDatum datum = null;
+ while (values.hasNext()) {
+ CrawlDatum nextDatum = (CrawlDatum)values.next();
+ if (datum == null || nextDatum.getStatus() > datum.getStatus())
+ datum = nextDatum;
+ }
+ output.collect(key, datum);
+ }
+}
Added: incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDatum.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDatum.java?rev=170983&view=auto
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDatum.java (added)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/CrawlDatum.java Thu May 19 12:53:00 2005
@@ -0,0 +1,139 @@
+/**
+ * Copyright 2005 The Apache Software Foundation
+ *
+ * Licensed 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.io.*;
+import java.net.*;
+import java.util.*;
+
+import org.apache.nutch.io.*;
+import org.apache.nutch.util.*;
+
+/* The crawl state of a url. */
+public class CrawlDatum implements Writable, Cloneable {
+ private final static byte CUR_VERSION = 0;
+
+ public static final byte STATUS_DB_UNFETCHED = 0;
+ public static final byte STATUS_DB_FETCHED = 1;
+ public static final byte STATUS_LINKED = 2;
+ public static final byte STATUS_FETCHER_SUCCESS = 3;
+ public static final byte STATUS_FETCHER_FAIL_TEMP = 4;
+ public static final byte STATUS_FETCHER_FAIL_PERM = 5;
+
+ private byte status;
+ private long nextFetch = System.currentTimeMillis();
+ private byte retries;
+ private float fetchInterval;
+
+ public CrawlDatum() {}
+
+ public CrawlDatum(int status, float fetchInterval) {
+ this.status = (byte)status;
+ this.fetchInterval = fetchInterval;
+ }
+
+ //
+ // accessor methods
+ //
+
+ public byte getStatus() { return status; }
+ public void setStatus(byte status) { this.status = (byte)status; }
+
+ public long getNextFetchTime() { return nextFetch; }
+ public void setNextFetchTime(long nextFetch) { this.nextFetch = nextFetch; }
+
+ public byte getRetriesSinceFetch() { return retries; }
+ public void setRetriesSinceFetch(int retries) {this.retries = (byte)retries;}
+
+ public float getFetchInterval() { return fetchInterval; }
+ public void setFetchInterval(float fetchInterval) {
+ this.fetchInterval = fetchInterval;
+ }
+
+ //
+ // writable methods
+ //
+
+ public void readFields(DataInput in) throws IOException {
+ byte version = in.readByte(); // read version
+ if (version > CUR_VERSION) // check version
+ throw new VersionMismatchException(CUR_VERSION, version);
+
+ status = in.readByte();
+ nextFetch = in.readLong();
+ retries = in.readByte();
+ fetchInterval = in.readFloat();
+ }
+
+ public void write(DataOutput out) throws IOException {
+ out.writeByte(CUR_VERSION); // store current version
+ out.write(status);
+ out.writeLong(nextFetch);
+ out.write(retries);
+ out.writeFloat(fetchInterval);
+ }
+
+ /** Copy the contents of another instance into this instance. */
+ public void set(CrawlDatum that) {
+ this.status = that.status;
+ this.nextFetch = that.nextFetch;
+ this.retries = that.retries;
+ this.fetchInterval = that.fetchInterval;
+ }
+
+ //
+ // basic methods
+ //
+
+ public String toString() {
+ StringBuffer buf = new StringBuffer();
+ buf.append("Version: " + CUR_VERSION + "\n");
+ buf.append("Status: " + getStatus() + "\n");
+ buf.append("Next fetch: " + new Date(getNextFetchTime()) + "\n");
+ buf.append("Retries since fetch: " + getRetriesSinceFetch() + "\n");
+ buf.append("Retry interval: " + getFetchInterval() + " days\n");
+ return buf.toString();
+ }
+
+ public boolean equals(Object o) {
+ if (!(o instanceof CrawlDatum))
+ return false;
+ CrawlDatum other = (CrawlDatum)o;
+ return
+ (this.status == other.status) &&
+ (this.nextFetch == other.nextFetch) &&
+ (this.retries == other.retries) &&
+ (this.fetchInterval == other.fetchInterval);
+ }
+
+ public int hashCode() {
+ return
+ status ^
+ ((int)nextFetch) ^
+ retries ^
+ Float.floatToIntBits(fetchInterval);
+ }
+
+ public Object clone() {
+ try {
+ return super.clone();
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+}
Added: incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/DmozParser.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/DmozParser.java?rev=170983&view=auto
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/DmozParser.java (added)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/DmozParser.java Thu May 19 12:53:00 2005
@@ -0,0 +1,384 @@
+/**
+ * Copyright 2005 The Apache Software Foundation
+ *
+ * Licensed 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.io.*;
+import java.net.*;
+import java.util.*;
+import java.util.logging.*;
+import java.net.MalformedURLException;
+import java.util.regex.*;
+
+import javax.xml.parsers.*;
+import org.xml.sax.*;
+import org.xml.sax.helpers.*;
+import org.apache.xerces.util.XMLChar;
+
+import org.apache.nutch.io.*;
+import org.apache.nutch.fs.*;
+import org.apache.nutch.net.*;
+import org.apache.nutch.util.*;
+import org.apache.nutch.util.NutchConf;
+
+/** Utility that converts DMOZ RDF into a flat file of URLs to be injected. */
+public class DmozParser {
+ public static final Logger LOG = LogFormatter.getLogger("org.apache.nutch.crawl.DmozParser");
+
+ long pages = 0;
+
+ /**
+ * This filter fixes characters that might offend our parser.
+ * This lets us be tolerant of errors that might appear in the input XML.
+ */
+ private static class XMLCharFilter extends FilterReader {
+ private boolean lastBad = false;
+
+ public XMLCharFilter(Reader reader) {
+ super(reader);
+ }
+
+ public int read() throws IOException {
+ int c = in.read();
+ int value = c;
+ if (c != -1 && !(XMLChar.isValid(c))) // fix invalid characters
+ value = 'X';
+ else if (lastBad && c == '<') { // fix mis-matched brackets
+ in.mark(1);
+ if (in.read() != '/')
+ value = 'X';
+ in.reset();
+ }
+ lastBad = (c == 65533);
+
+ return value;
+ }
+
+ public int read(char[] cbuf, int off, int len)
+ throws IOException {
+ int n = in.read(cbuf, off, len);
+ if (n != -1) {
+ for (int i = 0; i < n; i++) {
+ char c = cbuf[off+i];
+ char value = c;
+ if (!(XMLChar.isValid(c))) // fix invalid characters
+ value = 'X';
+ else if (lastBad && c == '<') { // fix mis-matched brackets
+ if (i != n-1 && cbuf[off+i+1] != '/')
+ value = 'X';
+ }
+ lastBad = (c == 65533);
+ cbuf[off+i] = value;
+ }
+ }
+ return n;
+ }
+ }
+
+
+ /**
+ * The RDFProcessor receives tag messages during a parse
+ * of RDF XML data. We build whatever structures we need
+ * from these messages.
+ */
+ private class RDFProcessor extends DefaultHandler {
+ String curURL = null, curSection = null;
+ boolean titlePending = false, descPending = false, insideAdultSection = false;
+ Pattern topicPattern = null;
+ StringBuffer title = new StringBuffer(), desc = new StringBuffer();
+ XMLReader reader;
+ int subsetDenom;
+ int hashSkew;
+ boolean includeAdult;
+ Locator location;
+
+ /**
+ * Pass in an XMLReader, plus a flag as to whether we
+ * should include adult material.
+ */
+ public RDFProcessor(XMLReader reader, int subsetDenom, boolean includeAdult, int skew, Pattern topicPattern) throws IOException {
+ this.reader = reader;
+ this.subsetDenom = subsetDenom;
+ this.includeAdult = includeAdult;
+ this.topicPattern = topicPattern;
+
+ this.hashSkew = skew != 0 ? skew : new Random().nextInt();
+ }
+
+ //
+ // Interface ContentHandler
+ //
+
+ /**
+ * Start of an XML elt
+ */
+ public void startElement(String namespaceURI, String localName, String qName, Attributes atts) throws SAXException {
+ if ("Topic".equals(qName)) {
+ curSection = atts.getValue("r:id");
+ } else if ("ExternalPage".equals(qName)) {
+ // Porn filter
+ if ((! includeAdult) && curSection.startsWith("Top/Adult")) {
+ return;
+ }
+
+ if (topicPattern != null && !topicPattern.matcher(curSection).matches()) {
+ return;
+ }
+
+ // Subset denominator filter.
+ // Only emit with a chance of 1/denominator.
+ String url = atts.getValue("about");
+ int hashValue = MD5Hash.digest(url).hashCode();
+ hashValue = Math.abs(hashValue ^ hashSkew);
+ if ((hashValue % subsetDenom) != 0) {
+ return;
+ }
+
+ // We actually claim the URL!
+ curURL = url;
+ } else if (curURL != null && "d:Title".equals(qName)) {
+ titlePending = true;
+ } else if (curURL != null && "d:Description".equals(qName)) {
+ descPending = true;
+ }
+ }
+
+ /**
+ * The contents of an XML elt
+ */
+ public void characters(char ch[], int start, int length) {
+ if (titlePending) {
+ title.append(ch, start, length);
+ } else if (descPending) {
+ desc.append(ch, start, length);
+ }
+ }
+
+ /**
+ * Termination of XML elt
+ */
+ public void endElement(String namespaceURI, String localName, String qName)
+ throws SAXException {
+ if (curURL != null) {
+ if ("ExternalPage".equals(qName)) {
+ //
+ // Inc the number of pages, insert the page, and
+ // possibly print status.
+ //
+ System.out.println(curURL);
+ pages++;
+
+ //
+ // Clear out the link text. This is what
+ // you would use for adding to the linkdb.
+ //
+ if (title.length() > 0) {
+ title.delete(0, title.length());
+ }
+ if (desc.length() > 0) {
+ desc.delete(0, desc.length());
+ }
+
+ // Null out the URL.
+ curURL = null;
+ } else if ("d:Title".equals(qName)) {
+ titlePending = false;
+ } else if ("d:Description".equals(qName)) {
+ descPending = false;
+ }
+ }
+ }
+
+ /**
+ * When parsing begins
+ */
+ public void startDocument() {
+ LOG.info("Begin parse");
+ }
+
+ /**
+ * When parsing ends
+ */
+ public void endDocument() {
+ LOG.info("Completed parse. Found " + pages + " pages.");
+ }
+
+ /**
+ * From time to time the Parser will set the "current location"
+ * by calling this function. It's useful for emitting locations
+ * for error messages.
+ */
+ public void setDocumentLocator(Locator locator) {
+ location = locator;
+ }
+
+
+ //
+ // Interface ErrorHandler
+ //
+
+ /**
+ * Emit the exception message
+ */
+ public void error(SAXParseException spe) {
+ LOG.severe("Error: " + spe.toString() + ": " + spe.getMessage());
+ spe.printStackTrace(System.err);
+ }
+
+ /**
+ * Emit the exception message, with line numbers
+ */
+ public void fatalError(SAXParseException spe) {
+ LOG.severe("Fatal err: " + spe.toString() + ": " + spe.getMessage());
+ LOG.severe("Last known line is " + location.getLineNumber() + ", column " + location.getColumnNumber());
+ spe.printStackTrace(System.err);
+ }
+
+ /**
+ * Emit exception warning message
+ */
+ public void warning(SAXParseException spe) {
+ LOG.warning("Warning: " + spe.toString() + ": " + spe.getMessage());
+ spe.printStackTrace(System.err);
+ }
+ }
+
+ /**
+ * Iterate through all the items in this structured DMOZ file.
+ * Add each URL to the web db.
+ */
+ public void parseDmozFile(File dmozFile, int subsetDenom,
+ boolean includeAdult,
+ int skew,
+ Pattern topicPattern)
+
+ throws IOException, SAXException, ParserConfigurationException {
+
+ SAXParserFactory parserFactory = SAXParserFactory.newInstance();
+ SAXParser parser = parserFactory.newSAXParser();
+ XMLReader reader = parser.getXMLReader();
+
+ // Create our own processor to receive SAX events
+ RDFProcessor rp =
+ new RDFProcessor(reader, subsetDenom, includeAdult,
+ skew, topicPattern);
+ reader.setContentHandler(rp);
+ reader.setErrorHandler(rp);
+ LOG.info("skew = " + rp.hashSkew);
+
+ //
+ // Open filtered text stream. The UTF8Filter makes sure that
+ // only appropriate XML-approved UTF8 characters are received.
+ // Any non-conforming characters are silently skipped.
+ //
+ XMLCharFilter in = new XMLCharFilter(new BufferedReader(new InputStreamReader(new BufferedInputStream(new FileInputStream(dmozFile)), "UTF-8")));
+ try {
+ InputSource is = new InputSource(in);
+ reader.parse(is);
+ } catch (Exception e) {
+ LOG.severe(e.toString());
+ e.printStackTrace(System.err);
+ System.exit(0);
+ } finally {
+ in.close();
+ }
+ }
+
+ private static void addTopicsFromFile(String topicFile, Vector topics)
+ throws IOException {
+ BufferedReader in = null;
+ try {
+ in = new BufferedReader(new InputStreamReader(new FileInputStream(topicFile), "UTF-8"));
+ String line = null;
+ while ((line = in.readLine()) != null) {
+ topics.addElement(new String(line));
+ }
+ }
+ catch (Exception e) {
+ LOG.severe(e.toString());
+ e.printStackTrace(System.out);
+ System.exit(0);
+ } finally {
+ in.close();
+ }
+ }
+
+ /**
+ * Command-line access. User may add URLs via a flat text file
+ * or the structured DMOZ file. By default, we ignore Adult
+ * material (as categorized by DMOZ).
+ */
+ public static void main(String argv[]) throws Exception {
+ if (argv.length < 1) {
+ System.err.println("Usage: DmozParser <dmoz_file> [-subset <subsetDenominator>] [-includeAdultMaterial] [-skew skew] [-topicFile <topic list file>] [-topic <topic> [-topic <topic> [...]]]");
+ return;
+ }
+
+ //
+ // Parse the command line, figure out what kind of
+ // URL file we need to load
+ //
+ int subsetDenom = 1;
+ int skew = 0;
+ String dmozFile = argv[0];
+ boolean includeAdult = false;
+ Pattern topicPattern = null;
+ Vector topics = new Vector();
+
+ NutchFileSystem nfs = NutchFileSystem.get();
+ try {
+ for (int i = 1; i < argv.length; i++) {
+ if ("-includeAdultMaterial".equals(argv[i])) {
+ includeAdult = true;
+ } else if ("-subset".equals(argv[i])) {
+ subsetDenom = Integer.parseInt(argv[i+1]);
+ i++;
+ } else if ("-topic".equals(argv[i])) {
+ topics.addElement(argv[i+1]);
+ i++;
+ } else if ("-topicFile".equals(argv[i])) {
+ addTopicsFromFile(argv[i+1], topics);
+ i++;
+ } else if ("-skew".equals(argv[i])) {
+ skew = Integer.parseInt(argv[i+1]);
+ i++;
+ }
+ }
+
+ DmozParser parser = new DmozParser();
+
+ if (!topics.isEmpty()) {
+ String regExp = new String("^(");
+ int j = 0;
+ for ( ; j < topics.size() - 1; ++j) {
+ regExp = regExp.concat((String) topics.get(j));
+ regExp = regExp.concat("|");
+ }
+ regExp = regExp.concat((String) topics.get(j));
+ regExp = regExp.concat(").*");
+ LOG.info("Topic selection pattern = " + regExp);
+ topicPattern = Pattern.compile(regExp);
+ }
+
+ parser.parseDmozFile(new File(dmozFile), subsetDenom,
+ includeAdult, skew, topicPattern);
+
+ } finally {
+ nfs.close();
+ }
+ }
+
+}
Copied: incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/Injector.java (from r170616, incubator/nutch/branches/mapred/src/java/org/apache/nutch/db/WebDBInjector.java)
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/Injector.java?p2=incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/Injector.java&p1=incubator/nutch/branches/mapred/src/java/org/apache/nutch/db/WebDBInjector.java&r1=170616&r2=170983&rev=170983&view=diff
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/db/WebDBInjector.java (original)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/Injector.java Thu May 19 12:53:00 2005
@@ -14,548 +14,102 @@
* limitations under the License.
*/
-package org.apache.nutch.db;
+package org.apache.nutch.crawl;
import java.io.*;
-import java.net.*;
import java.util.*;
import java.util.logging.*;
-import java.net.MalformedURLException;
-import java.util.regex.*;
-
-import javax.xml.parsers.*;
-import org.xml.sax.*;
-import org.xml.sax.helpers.*;
-import org.apache.xerces.util.XMLChar;
import org.apache.nutch.io.*;
import org.apache.nutch.fs.*;
import org.apache.nutch.net.*;
import org.apache.nutch.util.*;
-import org.apache.nutch.pagedb.*;
-import org.apache.nutch.linkdb.*;
-import org.apache.nutch.util.NutchConf;
-
-/*********************************************
- * This class takes a flat file of URLs and adds
- * them as entries into a pagedb. Useful for
- * bootstrapping the system.
- *
- * @author Mike Cafarella
- * @author Doug Cutting
- *********************************************/
-public class WebDBInjector {
- private static final String DMOZ_PAGENAME = "http://www.dmoz.org/";
-
- private static final byte DEFAULT_INTERVAL =
- (byte)NutchConf.get().getInt("db.default.fetch.interval", 30);
-
- private static final float NEW_INJECTED_PAGE_SCORE =
- NutchConf.get().getFloat("db.score.injected", 2.0f);
-
- public static final Logger LOG = LogFormatter.getLogger("org.apache.nutch.db.WebDBInjector");
-
- /**
- * This filter fixes characters that might offend our parser.
- * This lets us be tolerant of errors that might appear in the input XML.
- */
- private static class XMLCharFilter extends FilterReader {
- private boolean lastBad = false;
-
- public XMLCharFilter(Reader reader) {
- super(reader);
- }
-
- public int read() throws IOException {
- int c = in.read();
- int value = c;
- if (c != -1 && !(XMLChar.isValid(c))) // fix invalid characters
- value = 'X';
- else if (lastBad && c == '<') { // fix mis-matched brackets
- in.mark(1);
- if (in.read() != '/')
- value = 'X';
- in.reset();
- }
- lastBad = (c == 65533);
-
- return value;
- }
-
- public int read(char[] cbuf, int off, int len)
- throws IOException {
- int n = in.read(cbuf, off, len);
- if (n != -1) {
- for (int i = 0; i < n; i++) {
- char c = cbuf[off+i];
- char value = c;
- if (!(XMLChar.isValid(c))) // fix invalid characters
- value = 'X';
- else if (lastBad && c == '<') { // fix mis-matched brackets
- if (i != n-1 && cbuf[off+i+1] != '/')
- value = 'X';
- }
- lastBad = (c == 65533);
- cbuf[off+i] = value;
- }
- }
- return n;
- }
- }
-
-
- /**
- * The RDFProcessor receives tag messages during a parse
- * of RDF XML data. We build whatever structures we need
- * from these messages.
- */
- class RDFProcessor extends DefaultHandler {
- String curURL = null, curSection = null;
- boolean titlePending = false, descPending = false, insideAdultSection = false;
- Pattern topicPattern = null;
- StringBuffer title = new StringBuffer(), desc = new StringBuffer();
- XMLReader reader;
- int subsetDenom;
- int hashSkew;
- boolean includeAdult, includeDmozDesc;
- MD5Hash srcDmozID;
- long srcDmozDomainID;
- Locator location;
-
- /**
- * Pass in an XMLReader, plus a flag as to whether we
- * should include adult material.
- */
- public RDFProcessor(XMLReader reader, int subsetDenom, boolean includeAdult, boolean includeDmozDesc, int skew, Pattern topicPattern) throws IOException {
- this.reader = reader;
- this.subsetDenom = subsetDenom;
- this.includeAdult = includeAdult;
- this.includeDmozDesc = includeDmozDesc;
- this.topicPattern = topicPattern;
-
- // We create a Page entry for the "Dmoz" page, from
- // which all descriptive links originate. The name
- // of this page is always the same, stored in
- // DMOZ_PAGENAME. The MD5 is generated over the current
- // timestamp. Until this page is deleted, the descriptive
- // links will always be kept.
- //
- // If the DMOZ page is updated with new content, you
- // *could* update these links, if you really wanted to.
- // Just run inject again! This will replace the old
- // Dmoz Page, because we always keep the same name.
- // That obsolete Page will be deleted, and all its
- // outlinks (the descriptive ones) garbage-collected.
- //
- // Then we just proceed to add the new descriptive
- // links, with the brand-new page's src MD5.
- //
- this.srcDmozID = MD5Hash.digest(DMOZ_PAGENAME + "_" + nextFetch);
- Page dmozPage = new Page(DMOZ_PAGENAME, srcDmozID);
- dmozPage.setNextFetchTime(Long.MAX_VALUE);
- dbWriter.addPageIfNotPresent(dmozPage);
-
- this.srcDmozDomainID = MD5Hash.digest(new URL(DMOZ_PAGENAME).getHost()).halfDigest();
-
- this.hashSkew = skew != 0 ? skew : new Random().nextInt();
- }
-
- //
- // Interface ContentHandler
- //
-
- /**
- * Start of an XML elt
- */
- public void startElement(String namespaceURI, String localName, String qName, Attributes atts) throws SAXException {
- if ("Topic".equals(qName)) {
- curSection = atts.getValue("r:id");
- } else if ("ExternalPage".equals(qName)) {
- // Porn filter
- if ((! includeAdult) && curSection.startsWith("Top/Adult")) {
- return;
- }
-
- if (topicPattern != null && !topicPattern.matcher(curSection).matches()) {
- return;
- }
-
- // Subset denominator filter.
- // Only emit with a chance of 1/denominator.
- String url = atts.getValue("about");
- int hashValue = MD5Hash.digest(url).hashCode();
- hashValue = Math.abs(hashValue ^ hashSkew);
- if ((hashValue % subsetDenom) != 0) {
- return;
- }
-
- // We actually claim the URL!
- curURL = url;
- } else if (curURL != null && "d:Title".equals(qName)) {
- titlePending = true;
- } else if (curURL != null && "d:Description".equals(qName)) {
- descPending = true;
- }
- }
-
- /**
- * The contents of an XML elt
- */
- public void characters(char ch[], int start, int length) {
- if (titlePending) {
- title.append(ch, start, length);
- } else if (descPending) {
- desc.append(ch, start, length);
- }
- }
-
- /**
- * Termination of XML elt
- */
- public void endElement(String namespaceURI, String localName, String qName) throws SAXException {
- if (curURL != null) {
- if ("ExternalPage".equals(qName)) {
- //
- // Inc the number of pages, insert the page, and
- // possibly print status.
- //
- try {
- // First, manufacture the Page entry for the
- // given DMOZ listing.
- if (addPage(curURL)) {
-
- // Second, add a link from the DMOZ page TO the
- // just-added target Page. The anchor text should
- // be the merged Title and Desc that we get from
- // the DMOZ listing. For testing reasons, the
- // caller may choose to disallow this.
- if (includeDmozDesc) {
- String fullDesc = title + " " + desc;
- Link descLink = new Link(srcDmozID, srcDmozDomainID, curURL, fullDesc);
- dbWriter.addLink(descLink);
- }
- pages++;
- }
-
- } catch (MalformedURLException e) {
- LOG.fine("skipping " + curURL + ":" + e);
- } catch (IOException ie) {
- LOG.severe("problem adding url " + curURL + ": " + ie);
- }
- printStatusBar(2000, 50000);
-
- //
- // Clear out the link text. This is what
- // you would use for adding to the linkdb.
- //
- if (title.length() > 0) {
- title.delete(0, title.length());
- }
- if (desc.length() > 0) {
- desc.delete(0, desc.length());
- }
-
- // Null out the URL.
- curURL = null;
- } else if ("d:Title".equals(qName)) {
- titlePending = false;
- } else if ("d:Description".equals(qName)) {
- descPending = false;
- }
- }
- }
-
- /**
- * When parsing begins
- */
- public void startDocument() {
- LOG.info("Begin parse");
- }
-
- /**
- * When parsing ends
- */
- public void endDocument() {
- LOG.info("Completed parse. Added " + pages + " pages.");
- }
-
- /**
- * From time to time the Parser will set the "current location"
- * by calling this function. It's useful for emitting locations
- * for error messages.
- */
- public void setDocumentLocator(Locator locator) {
- location = locator;
- }
-
-
- //
- // Interface ErrorHandler
- //
-
- /**
- * Emit the exception message
- */
- public void error(SAXParseException spe) {
- LOG.severe("Error: " + spe.toString() + ": " + spe.getMessage());
- spe.printStackTrace(System.out);
- }
-
- /**
- * Emit the exception message, with line numbers
- */
- public void fatalError(SAXParseException spe) {
- LOG.severe("Fatal error: " + spe.toString() + ": " + spe.getMessage());
- LOG.severe("Last known line is " + location.getLineNumber() + ", column " + location.getColumnNumber());
- spe.printStackTrace(System.out);
- }
-
- /**
- * Emit exception warning message
- */
- public void warning(SAXParseException spe) {
- LOG.warning("Warning: " + spe.toString() + ": " + spe.getMessage());
- spe.printStackTrace(System.out);
- }
- }
-
- private IWebDBWriter dbWriter;
-
- /**
- * WebDBInjector takes a reference to a WebDBWriter that it should add to.
- */
- public WebDBInjector(IWebDBWriter dbWriter) {
- this.dbWriter = dbWriter;
- }
-
- /**
- * Close dbWriter and save changes
- */
- public void close() throws IOException {
- dbWriter.close();
- }
+import org.apache.nutch.mapred.*;
- /**
- * Utility to present small status bar
- */
- public void printStatusBar(int small, int big){
- if ((pages % small ) == 0) {
- System.out.print(".");
- }
- if ((pages % big ) == 0) {
- printStatus();
- }
- }
+/** This class takes a flat file of URLs and adds them to the of pages to be
+ * crawled. Useful for bootstrapping the system. */
+public class Injector extends NutchConfigured {
+ public static final Logger LOG =
+ LogFormatter.getLogger("org.apache.nutch.crawl.Injector");
- long startTime = System.currentTimeMillis();
- long pages = 0;
- long nextFetch = System.currentTimeMillis();
-
- /**
- * Utility to present performance stats
- */
- public void printStatus(){
- long elapsed = (System.currentTimeMillis() - this.startTime);
- if ( this.pages == 0) {
- } else {
- LOG.info("\t" + this.pages + "\t" +
- (int)((1000 * pages)/elapsed) + " pages/second\t" );
- }
- }
- /**
- * Iterate through all the items in this flat text file and
- * add them to the db.
- */
- public void injectURLFile(File urlList) throws IOException {
- nextFetch = urlList.lastModified();
- BufferedReader reader = new BufferedReader(new FileReader(urlList));
- try {
- String curStr = null;
- LOG.info("Starting URL processing");
- while ((curStr = reader.readLine()) != null) {
- String url = curStr.trim();
- if (addPage(url))
- this.pages++;
- printStatusBar(2000,50000);
- }
- LOG.info("Added " + pages + " pages");
- } catch (Exception e) {
- LOG.severe("error while injecting:" + e);
- e.printStackTrace();
- } finally {
- reader.close();
- }
- }
+ /** Normalize and filter injected urls. */
+ public static class InjectMapper implements Mapper {
+ private UrlNormalizer urlNormalizer = UrlNormalizerFactory.getNormalizer();
+ private float interval;
- /**
- * Iterate through all the items in this structured DMOZ file.
- * Add each URL to the web db.
- */
- public void injectDmozFile(File dmozFile, int subsetDenom, boolean includeAdult, boolean includeDmozDesc, int skew, Pattern topicPattern) throws IOException, SAXException, ParserConfigurationException {
- nextFetch = dmozFile.lastModified();
-
- SAXParserFactory parserFactory = SAXParserFactory.newInstance();
- SAXParser parser = parserFactory.newSAXParser();
- XMLReader reader = parser.getXMLReader();
-
- // Create our own processor to receive SAX events
- RDFProcessor rp =
- new RDFProcessor(reader, subsetDenom, includeAdult, includeDmozDesc, skew, topicPattern);
- reader.setContentHandler(rp);
- reader.setErrorHandler(rp);
- LOG.info("skew = " + rp.hashSkew);
-
- //
- // Open filtered text stream. The UTF8Filter makes sure that
- // only appropriate XML-approved UTF8 characters are received.
- // Any non-conforming characters are silently skipped.
- //
- XMLCharFilter in = new XMLCharFilter(new BufferedReader(new InputStreamReader(new BufferedInputStream(new FileInputStream(dmozFile)), "UTF-8")));
- try {
- InputSource is = new InputSource(in);
- reader.parse(is);
- } catch (Exception e) {
- LOG.severe(e.toString());
- e.printStackTrace(System.out);
- System.exit(0);
- } finally {
- in.close();
- }
+ public void configure(JobConf job) {
+ interval = job.getFloat("db.default.fetch.interval", 30f);
}
- /**
- * Add one page to WebDB. Changes are not saved to the db until
- * the <code>close()</code> is invoked. URLs are checked with the
- * URLFilter, and only those that pass are added.
- * @param url URL to be added
- * @return true on success, false otherwise (e.g. filtered out or
- * bad URL syntax).
- */
- public boolean addPage(String url) throws IOException {
+ public void map(WritableComparable key, Writable val,
+ OutputCollector output) throws IOException {
+ UTF8 value = (UTF8)val;
+ String url = value.toString(); // value is line of text
try {
- url = URLFilters.filter(url);
- } catch (URLFilterException e) {
- throw new IOException(e.getMessage());
- }
- if (url != null) {
- try {
- Page page = new Page(url, NEW_INJECTED_PAGE_SCORE, nextFetch);
- dbWriter.addPageIfNotPresent(page);
- return true;
- } catch (MalformedURLException e) {
- LOG.warning("bad url: "+url);
- }
- }
- return false;
- }
+ url = urlNormalizer.normalize(url); // normalize the url
+ url = URLFilters.filter(url); // filter the url
+ } catch (Exception e) {
+ LOG.warning("Skipping " +url+":"+e);
+ url = null;
+ }
+ if (url != null) { // if it passes
+ value.set(url); // collect it
+ output.collect(value, new CrawlDatum(CrawlDatum.STATUS_DB_UNFETCHED,
+ interval));
+ }
+ }
+ }
+
+ /** Construct an Injector. */
+ public Injector(NutchConf conf) {
+ super(conf);
+ }
+
+ public void inject(File urlDir, File crawlDb) throws IOException {
+ File tempDir =
+ new File("inject-temp-"+
+ Integer.toString(new Random().nextInt(Integer.MAX_VALUE)));
+
+ // map text input file to a <url,CrawlDatum> file
+ JobConf sortJob = new JobConf(getConf());
+ sortJob.setInputDir(urlDir);
+ sortJob.setMapperClass(InjectMapper.class);
+
+ sortJob.setOutputDir(tempDir);
+ sortJob.setOutputFormat(OutputFormats.get("seq"));
+ sortJob.setOutputKeyClass(UTF8.class);
+ sortJob.setOutputValueClass(CrawlDatum.class);
+ JobClient.runJob(sortJob);
+
+ // merge with existing crawl db
+ File newCrawlDb =
+ new File(crawlDb,
+ Integer.toString(new Random().nextInt(Integer.MAX_VALUE)));
+ JobConf mergeJob = new JobConf(getConf());
+ mergeJob.addInputDir(tempDir);
+ mergeJob.addInputDir(crawlDb);
+ mergeJob.setInputFormat(InputFormats.get("seq"));
+ mergeJob.setInputKeyClass(UTF8.class);
+ mergeJob.setInputValueClass(CrawlDatum.class);
+
+ mergeJob.setPartitionerClass(CrawlDBPartitioner.class);
+ mergeJob.setReducerClass(CrawlDBReducer.class);
+
+ mergeJob.setOutputDir(newCrawlDb);
+ mergeJob.setOutputFormat(OutputFormats.get("seq"));
+ mergeJob.setOutputKeyClass(UTF8.class);
+ mergeJob.setOutputValueClass(CrawlDatum.class);
+
+ JobClient.runJob(mergeJob);
+
+ new JobClient(getConf()).getFs().delete(tempDir);
+
+ }
+
+ public static void main(String[] args) throws Exception {
+ Injector injector = new Injector(NutchConf.get());
+ injector.inject(new File(args[0]), new File(args[1]));
+ }
- private static void addTopicsFromFile(String topicFile, Vector topics) throws IOException {
- BufferedReader in = null;
- try {
- in = new BufferedReader(new InputStreamReader(new FileInputStream(topicFile), "UTF-8"));
- String line = null;
- while ((line = in.readLine()) != null) {
- topics.addElement(new String(line));
- }
- }
- catch (Exception e) {
- LOG.severe(e.toString());
- e.printStackTrace(System.out);
- System.exit(0);
- } finally {
- in.close();
- }
- }
-
-
- /**
- * Command-line access. User may add URLs via a flat text file
- * or the structured DMOZ file. By default, we ignore Adult
- * material (as categorized by DMOZ).
- */
- public static void main(String argv[]) throws Exception {
- if (argv.length < 3) {
- System.out.println("Usage: WebDBInjector (-local | -ndfs <namenode:port>) <db_dir> (-urlfile <url_file> | -dmozfile <dmoz_file>) [-subset <subsetDenominator>] [-includeAdultMaterial] [-skew skew] [-noDmozDesc] [-topicFile <topic list file>] [-topic <topic> [-topic <topic> [...]]]");
- return;
- }
-
- //
- // Parse the command line, figure out what kind of
- // URL file we need to load
- //
- int subsetDenom = 1;
- int skew = 0;
- String command = null, loadfile = null;
- boolean includeAdult = false, includeDmozDesc = true;
- Pattern topicPattern = null;
- Vector topics = new Vector();
-
- int i = 0;
- NutchFileSystem nfs = NutchFileSystem.parseArgs(argv, i);
- try {
- File root = new File(argv[i++]);
-
- for (; i < argv.length; i++) {
- if ("-urlfile".equals(argv[i]) ||
- "-dmozfile".equals(argv[i])) {
- command = argv[i];
- loadfile = argv[i+1];
- i++;
- } else if ("-includeAdultMaterial".equals(argv[i])) {
- includeAdult = true;
- } else if ("-noDmozDesc".equals(argv[i])) {
- includeDmozDesc = false;
- } else if ("-subset".equals(argv[i])) {
- subsetDenom = Integer.parseInt(argv[i+1]);
- i++;
- } else if ("-topic".equals(argv[i])) {
- topics.addElement(argv[i+1]);
- i++;
- } else if ("-topicFile".equals(argv[i])) {
- addTopicsFromFile(argv[i+1], topics);
- i++;
- } else if ("-skew".equals(argv[i])) {
- skew = Integer.parseInt(argv[i+1]);
- i++;
- }
- }
-
- //
- // Create the webdbWriter, the injector, and then inject the
- // right kind of URL file.
- //
- IWebDBWriter writer = new WebDBWriter(nfs, root);
- WebDBInjector injector = new WebDBInjector(writer);
- try {
- if ("-urlfile".equals(command)) {
- if (!topics.isEmpty()) {
- System.out.println("You can't select URLs based on a topic when usin a URL-file");
- }
- injector.injectURLFile(new File(loadfile));
- } else if ("-dmozfile".equals(command)) {
- if (!topics.isEmpty()) {
- String regExp = new String("^(");
- int j = 0;
- for ( ; j < topics.size() - 1; ++j) {
- regExp = regExp.concat((String) topics.get(j));
- regExp = regExp.concat("|");
- }
- regExp = regExp.concat((String) topics.get(j));
- regExp = regExp.concat(").*");
- LOG.info("Topic selection pattern = " + regExp);
- topicPattern = Pattern.compile(regExp);
- }
- injector.injectDmozFile(new File(loadfile), subsetDenom, includeAdult, includeDmozDesc, skew, topicPattern);
- } else {
- System.out.println("No command indicated.");
- return;
- }
- } finally {
- injector.close();
- }
- } finally {
- nfs.close();
- }
- }
}
Added: incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/package.html
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/package.html?rev=170983&view=auto
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/package.html (added)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/crawl/package.html Thu May 19 12:53:00 2005
@@ -0,0 +1,5 @@
+<html>
+<body>
+Crawl control code.
+</body>
+</html>
Modified: incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/InputFormatBase.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/InputFormatBase.java?rev=170983&r1=170982&r2=170983&view=diff
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/InputFormatBase.java (original)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/InputFormatBase.java Thu May 19 12:53:00 2005
@@ -19,6 +19,7 @@
import java.io.IOException;
import java.io.File;
+import java.util.Arrays;
import java.util.ArrayList;
import org.apache.nutch.fs.NutchFileSystem;
@@ -38,11 +39,16 @@
* expression.*/
protected File[] listFiles(NutchFileSystem fs, JobConf job)
throws IOException {
- File[] files = fs.listFiles(job.getInputDir());
- if (files == null) {
- throw new IOException("No input files in: "+job.getInputDir());
+ File[] dirs = job.getInputDirs();
+ ArrayList files = new ArrayList();
+ for (int i = 0; i < dirs.length; i++) {
+ files.addAll(Arrays.asList(fs.listFiles(dirs[i])));
}
- return files;
+
+ if (files.size() == 0) {
+ throw new IOException("No input files in: "+job.getInputDirs());
+ }
+ return (File[])files.toArray(new File[files.size()]);
}
/** Splits files returned by {#listFiles(NutchFileSystem,JobConf) when
Modified: incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobClient.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobClient.java?rev=170983&r1=170982&r2=170983&view=diff
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobClient.java (original)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobClient.java Thu May 19 12:53:00 2005
@@ -164,8 +164,8 @@
/**
* Build a job client, connect to the default job tracker
*/
- public JobClient() throws IOException {
- this(JobTracker.getDefaultAddress());
+ public JobClient(NutchConf conf) throws IOException {
+ this(JobTracker.getAddress(conf));
}
/**
@@ -226,8 +226,12 @@
String originalJarPath = job.getJar();
- // Modify the job file, write to JobTracker's fs
- job.setJar(submitJarFile.toString());
+ if (originalJarPath != null) { // Copy jar to JobTracker's fs
+ job.setJar(submitJarFile.toString());
+ getFs().copyFromLocalFile(new File(originalJarPath), submitJarFile);
+ }
+
+ // Write job file to JobTracker's fs
NFSDataOutputStream out =
new NFSDataOutputStream(getFs().create(submitJobFile));
try {
@@ -235,8 +239,6 @@
} finally {
out.close();
}
- // Copy jar to JobTracker's fs
- getFs().copyFromLocalFile(new File(originalJarPath), submitJarFile);
//
// Now, actually submit the job (using the submit name)
@@ -265,7 +267,7 @@
/** Utility that submits a job, then polls for progress until the job is
* complete. */
public static void runJob(JobConf job) throws IOException {
- JobClient jc = new JobClient();
+ JobClient jc = new JobClient(job);
boolean error = true;
RunningJob running = null;
try {
@@ -327,7 +329,7 @@
}
// Submit the request
- JobClient jc = new JobClient();
+ JobClient jc = new JobClient(NutchConf.get());
try {
if (submitJobFile != null) {
RunningJob job = jc.submitJob(submitJobFile);
Modified: incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobConf.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobConf.java?rev=170983&r1=170982&r2=170983&view=diff
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobConf.java (original)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobConf.java Thu May 19 12:53:00 2005
@@ -25,7 +25,10 @@
import java.util.Properties;
import java.util.jar.JarFile;
import java.util.jar.JarEntry;
+import java.util.StringTokenizer;
+import java.util.ArrayList;
import java.util.List;
+import java.util.Collections;
import org.apache.nutch.fs.NutchFileSystem;
import org.apache.nutch.util.NutchConf;
@@ -46,11 +49,15 @@
* of input files, and where the output files should be written. */
public class JobConf extends NutchConf {
- /** Construct a map/reduce configuration. */
- public JobConf() {
- super();
+ /** Construct a map/reduce job configuration.
+ *
+ * @param confg a NutchConf whose settings will be inherited.
+ */
+ public JobConf(NutchConf conf) {
+ super(conf);
}
+
/** Construct a map/reduce configuration.
*
* @param config a NutchConf-format XML job description file
@@ -68,15 +75,7 @@
addConfResource(config);
}
- public String getJar() {
- String defaultValue = "nutch.jar";
-
- URL jarUrl = get().getResource(defaultValue); // resolve default
- if (jarUrl != null && "file".equals(jarUrl.getProtocol()))
- defaultValue = jarUrl.getFile();
-
- return get("mapred.jar", defaultValue);
- }
+ public String getJar() { return get("mapred.jar"); }
public void setJar(String jar) { set("mapred.jar", jar); }
public static File getSystemDir() {
@@ -89,9 +88,22 @@
"/tmp/nutch/mapred/local"));
}
- public File getInputDir() { return new File(get("mapred.input.dir")); }
public void setInputDir(File dir) { set("mapred.input.dir", dir); }
+ public void addInputDir(File dir) {
+ String dirs = get("mapred.input.dir");
+ set("mapred.input.dir", dirs == null ? dir.toString() : dirs + "," + dir);
+ }
+ public File[] getInputDirs() {
+ String dirs = get("mapred.input.dir", "");
+ ArrayList list = Collections.list(new StringTokenizer(dirs, ","));
+ File[] result = new File[list.size()];
+ for (int i = 0; i < list.size(); i++) {
+ result[i] = new File((String)list.get(i));
+ }
+ return result;
+ }
+
public File getOutputDir() { return new File(get("mapred.output.dir")); }
public void setOutputDir(File dir) { set("mapred.output.dir", dir); }
@@ -120,7 +132,7 @@
return getClass("mapred.input.value.class", UTF8.class, Writable.class);
}
public void setInputValueClass(Class theClass) {
- setClass("mapred.input.value.class", theClass, WritableComparable.class);
+ setClass("mapred.input.value.class", theClass, Writable.class);
}
public Class getOutputKeyClass() {
@@ -146,7 +158,7 @@
return getClass("mapred.output.value.class", UTF8.class, Writable.class);
}
public void setOutputValueClass(Class theClass) {
- setClass("mapred.output.value.class", theClass, WritableComparable.class);
+ setClass("mapred.output.value.class", theClass, Writable.class);
}
public Class getMapperClass() {
@@ -196,8 +208,5 @@
return result;
}
- public static void main(String[] args) {
- System.out.println(new JobConf().getJar());
- }
}
Modified: incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobTracker.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobTracker.java?rev=170983&r1=170982&r2=170983&view=diff
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobTracker.java (original)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/JobTracker.java Thu May 19 12:53:00 2005
@@ -36,8 +36,8 @@
public static final Logger LOG = LogFormatter.getLogger("org.apache.nutch.mapred.JobTracker");
public static JobTracker tracker = null;
- public static void createTracker() throws IOException {
- createTracker(getDefaultAddress());
+ public static void createTracker(NutchConf conf) throws IOException {
+ createTracker(getAddress(conf));
}
public static void createTracker(InetSocketAddress addr) throws IOException {
tracker = new JobTracker(addr);
@@ -46,9 +46,9 @@
return tracker;
}
- public static InetSocketAddress getDefaultAddress() {
+ public static InetSocketAddress getAddress(NutchConf conf) {
String jobTrackerStr =
- NutchConf.get().get("mapred.job.tracker", "localhost:8012");
+ conf.get("mapred.job.tracker", "localhost:8012");
int colon = jobTrackerStr.indexOf(":");
if (colon < 0) {
throw new RuntimeException("Bad mapred.job.tracker: "+jobTrackerStr);
@@ -230,8 +230,8 @@
this.port = addr.getPort();
this.interTrackerServer = RPC.getServer(this, addr.getPort());
this.interTrackerServer.start();
- this.infoServer = new JobTrackerInfoServer(this, TRACKERINFO_PORT);
- this.infoServer.start();
+// this.infoServer = new JobTrackerInfoServer(this, TRACKERINFO_PORT);
+// this.infoServer.start();
this.startTime = System.currentTimeMillis();
@@ -903,7 +903,7 @@
System.exit(-1);
}
- JobTracker.createTracker();
+ JobTracker.createTracker(NutchConf.get());
JobTracker jt = JobTracker.getTracker();
jt.offerService();
}
Modified: incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskRunner.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskRunner.java?rev=170983&r1=170982&r2=170983&view=diff
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskRunner.java (original)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskRunner.java Thu May 19 12:53:00 2005
@@ -60,15 +60,40 @@
prepare();
+ String sep = System.getProperty("path.separator");
+ File workDir = new File(new File(t.getJobFile()).getParent(), "work");
+
+ StringBuffer classPath = new StringBuffer();
+ // start with same classpath as parent process
+ classPath.append(System.getProperty("java.class.path"));
+ classPath.append(sep);
+
+
+ JobConf job = new JobConf(t.getJobFile());
+ String jar = job.getJar();
+ if (jar != null) { // if jar exists, it into workDir
+ runChild(new String[] { "jar", "xf", jar}, workDir);
+ String[] libs = new File(workDir, "lib").list();
+ for (int i = 0; i < libs.length; i++) {
+ classPath.append(sep); // add libs from jar to classpath
+ classPath.append(libs[i]);
+ }
+ classPath.append(sep);
+ classPath.append(new File(workDir, "classes"));
+ classPath.append(sep);
+ classPath.append(workDir);
+ }
+
+ // run java
runChild(new String[] {
- "java", // run java
- //"-verbose",
- "-cp",
- new JobConf(t.getJobFile()).getJar(), // classpath is task jar
+ "java",
+ "-Xmx"+job.get("mapred.child.heap.size", "200m"),
+ "-cp", classPath.toString(),
TaskTracker.Child.class.getName(), // main is Child
tracker.taskReportPort+"", // pass umbilical port
t.getTaskId() // pass task identifier
- });
+ }, null);
+
} catch (IOException e) {
LOG.log(Level.WARNING, "Child Error", e);
} finally {
@@ -79,8 +104,8 @@
/**
* Run the child process
*/
- private void runChild(String[] args) throws IOException {
- this.process = Runtime.getRuntime().exec(args);
+ private void runChild(String[] args, File dir) throws IOException {
+ this.process = Runtime.getRuntime().exec(args, null, dir);
try {
logStream(process.getErrorStream()); // copy log output
logStream(process.getInputStream()); // normally empty
Modified: incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskTracker.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskTracker.java?rev=170983&r1=170982&r2=170983&view=diff
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskTracker.java (original)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskTracker.java Thu May 19 12:53:00 2005
@@ -64,7 +64,7 @@
* Start with the local machine name, and the default JobTracker
*/
public TaskTracker() throws IOException {
- this(JobTracker.getDefaultAddress());
+ this(JobTracker.getAddress(NutchConf.get()));
}
/**
@@ -306,15 +306,17 @@
JobConf jc = new JobConf(localJobFile);
String jarFile = jc.getJar();
- fs.copyToLocalFile(new File(jarFile), localJarFile);
- jc.setJar(localJarFile.getCanonicalPath());
-
- BufferedOutputStream out =
- new BufferedOutputStream(new FileOutputStream(localJobFile));
- try {
+ if (jarFile != null) {
+ fs.copyToLocalFile(new File(jarFile), localJarFile);
+ jc.setJar(localJarFile.getCanonicalPath());
+
+ BufferedOutputStream out =
+ new BufferedOutputStream(new FileOutputStream(localJobFile));
+ try {
jc.write(out);
- } finally {
+ } finally {
out.close();
+ }
}
}
Modified: incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/demo/Grep.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/demo/Grep.java?rev=170983&r1=170982&r2=170983&view=diff
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/demo/Grep.java (original)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/mapred/demo/Grep.java Thu May 19 12:53:00 2005
@@ -30,6 +30,8 @@
import org.apache.nutch.io.LongWritable;
import org.apache.nutch.io.WritableComparator;
+import org.apache.nutch.util.NutchConf;
+
import java.io.File;
import java.util.Random;
@@ -43,11 +45,13 @@
System.exit(-1);
}
+ NutchConf defaults = NutchConf.get();
+
File tempDir =
new File("grep-temp-"+
Integer.toString(new Random().nextInt(Integer.MAX_VALUE)));
- JobConf grepJob = new JobConf();
+ JobConf grepJob = new JobConf(defaults);
grepJob.setNumMapTasks(18);
grepJob.setInputDir(new File(args[0]));
@@ -68,7 +72,7 @@
JobClient.runJob(grepJob);
- JobConf sortJob = new JobConf();
+ JobConf sortJob = new JobConf(defaults);
sortJob.setNumMapTasks(6);
@@ -86,7 +90,7 @@
JobClient.runJob(sortJob);
- new JobClient().getFs().delete(tempDir);
+ new JobClient(defaults).getFs().delete(tempDir);
}
}
Modified: incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConf.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConf.java?rev=170983&r1=170982&r2=170983&view=diff
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConf.java (original)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConf.java Thu May 19 12:53:00 2005
@@ -47,7 +47,7 @@
/** Return the default configuration. */
public static NutchConf get() { return DEFAULT; }
- private List resourceNames = new ArrayList();
+ private ArrayList resourceNames = new ArrayList();
private Properties properties;
private ClassLoader classLoader = NutchConf.class.getClassLoader();
@@ -55,6 +55,13 @@
public NutchConf() {
resourceNames.add("nutch-default.xml");
resourceNames.add("nutch-site.xml");
+ }
+
+ /** A new configuration with the same settings as another. */
+ public NutchConf(NutchConf other) {
+ this.resourceNames = (ArrayList)other.resourceNames.clone();
+ if (other.properties != null)
+ this.properties = (Properties)other.properties.clone();
}
/** Adds a resource name to the chain of resources read. Such resources are
Added: incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConfigurable.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConfigurable.java?rev=170983&view=auto
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConfigurable.java (added)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConfigurable.java Thu May 19 12:53:00 2005
@@ -0,0 +1,27 @@
+/**
+ * Copyright 2005 The Apache Software Foundation
+ *
+ * Licensed 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.util;
+
+/** Something that may be configured with a {@link NutchConf}. */
+public interface NutchConfigurable {
+
+ /** Set the configuration to be used by this object. */
+ void setConf(NutchConf conf);
+
+ /** Return the configuration used by this object. */
+ NutchConf getConf();
+}
Added: incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConfigured.java
URL: http://svn.apache.org/viewcvs/incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConfigured.java?rev=170983&view=auto
==============================================================================
--- incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConfigured.java (added)
+++ incubator/nutch/branches/mapred/src/java/org/apache/nutch/util/NutchConfigured.java Thu May 19 12:53:00 2005
@@ -0,0 +1,39 @@
+/**
+ * Copyright 2005 The Apache Software Foundation
+ *
+ * Licensed 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.util;
+
+/** Base class for things that may be configured with a {@link NutchConf}. */
+public class NutchConfigured implements NutchConfigurable {
+
+ private NutchConf conf;
+
+ /** Construct a NutchConfigured. */
+ public NutchConfigured(NutchConf conf) {
+ setConf(conf);
+ }
+
+ // inherit javadoc
+ public void setConf(NutchConf conf) {
+ this.conf = conf;
+ }
+
+ // inherit javadoc
+ public NutchConf getConf() {
+ return conf;
+ }
+
+}