You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nutch.apache.org by ku...@apache.org on 2007/11/04 17:01:55 UTC
svn commit: r591793 - in /lucene/nutch/trunk: ./
src/java/org/apache/nutch/tools/arc/
Author: kubes
Date: Sun Nov 4 08:01:53 2007
New Revision: 591793
URL: http://svn.apache.org/viewvc?rev=591793&view=rev
Log:
NUTCH-565 - Arc File to Nutch Segments Converter. This tools allows the conversion of multiple .arc files, a format used by the internet archive and grub distributed crawler projects, into Nutch segments.
Added:
lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/
lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcInputFormat.java
lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcRecordReader.java
lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcSegmentCreator.java
Modified:
lucene/nutch/trunk/CHANGES.txt
Modified: lucene/nutch/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/lucene/nutch/trunk/CHANGES.txt?rev=591793&r1=591792&r2=591793&view=diff
==============================================================================
--- lucene/nutch/trunk/CHANGES.txt (original)
+++ lucene/nutch/trunk/CHANGES.txt Sun Nov 4 08:01:53 2007
@@ -156,6 +156,8 @@
53. NUTCH-552 - Upgrade Nutch to Hadoop 0.15.x. (kubes)
+54. NUTCH-565 - Arc File to Nutch Segments Converter. (kubes)
+
Release 0.9 - 2007-04-02
1. Changed log4j confiquration to log to stdout on commandline
Added: lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcInputFormat.java
URL: http://svn.apache.org/viewvc/lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcInputFormat.java?rev=591793&view=auto
==============================================================================
--- lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcInputFormat.java (added)
+++ lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcInputFormat.java Sun Nov 4 08:01:53 2007
@@ -0,0 +1,48 @@
+/**
+ * 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.tools.arc;
+
+import java.io.IOException;
+
+import org.apache.hadoop.mapred.FileInputFormat;
+import org.apache.hadoop.mapred.FileSplit;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapred.Reporter;
+
+/**
+ * A input format the reads arc files.
+ */
+public class ArcInputFormat
+ extends FileInputFormat {
+
+ /**
+ * Returns the <code>RecordReader</code> for reading the arc file.
+ *
+ * @param split The InputSplit of the arc file to process.
+ * @param job The job configuration.
+ * @param reporter The progress reporter.
+ */
+ public RecordReader getRecordReader(InputSplit split, JobConf job,
+ Reporter reporter)
+ throws IOException {
+ reporter.setStatus(split.toString());
+ return new ArcRecordReader(job, (FileSplit)split);
+ }
+
+}
Added: lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcRecordReader.java
URL: http://svn.apache.org/viewvc/lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcRecordReader.java?rev=591793&view=auto
==============================================================================
--- lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcRecordReader.java (added)
+++ lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcRecordReader.java Sun Nov 4 08:01:53 2007
@@ -0,0 +1,293 @@
+/**
+ * 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.tools.arc;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.zip.GZIPInputStream;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapred.FileSplit;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+
+/**
+ * <p>The <code>ArchRecordReader</code> class provides a record reader which
+ * reads records from arc files.</p>
+ *
+ * <p>Arc files are essentially tars of gzips. Each record in an arc file is
+ * a compressed gzip. Multiple records are concatenated together to form a
+ * complete arc. For more information on the arc file format see
+ * {@link http://www.archive.org/web/researcher/ArcFileFormat.php}.</p>
+ *
+ * <p>Arc files are used by the internet archive and grub projects.</p>
+ *
+ * @see http://www.archive.org/
+ * @see http://www.grub.org/
+ */
+public class ArcRecordReader
+ implements RecordReader {
+
+ public static final Log LOG = LogFactory.getLog(ArcRecordReader.class);
+
+ protected Configuration conf;
+ protected long splitStart = 0;
+ protected long pos = 0;
+ protected long splitEnd = 0;
+ protected long splitLen = 0;
+ protected long fileLen = 0;
+ protected FSDataInputStream in;
+
+ private static byte[] MAGIC = {(byte)0x1F, (byte)0x8B};
+
+ /**
+ * <p>Returns true if the byte array passed matches the gzip header magic
+ * number.</p>
+ *
+ * @param input The byte array to check.
+ *
+ * @return True if the byte array matches the gzip header magic number.
+ */
+ public static boolean isMagic(byte[] input) {
+
+ // check for null and incorrect length
+ if (input == null || input.length != MAGIC.length) {
+ return false;
+ }
+
+ // check byte by byte
+ for (int i = 0; i < MAGIC.length; i++) {
+ if (MAGIC[i] != input[i]) {
+ return false;
+ }
+ }
+
+ // must match
+ return true;
+ }
+
+ /**
+ * Constructor that sets the configuration and file split.
+ *
+ * @param conf The job configuration.
+ * @param split The file split to read from.
+ *
+ * @throws IOException If an IO error occurs while initializing file split.
+ */
+ public ArcRecordReader(Configuration conf, FileSplit split)
+ throws IOException {
+
+ Path path = split.getPath();
+ FileSystem fs = path.getFileSystem(conf);
+ fileLen = fs.getFileStatus(split.getPath()).getLen();
+ this.conf = conf;
+ this.in = fs.open(split.getPath());
+ this.splitStart = split.getStart();
+ this.splitEnd = splitStart + split.getLength();
+ this.splitLen = split.getLength();
+ in.seek(splitStart);
+ }
+
+ /**
+ * Closes the record reader resources.
+ */
+ public void close()
+ throws IOException {
+ this.in.close();
+ }
+
+ /**
+ * Creates a new instance of the <code>Text</code> object for the key.
+ */
+ public WritableComparable createKey() {
+ return (WritableComparable)ReflectionUtils.newInstance(Text.class, conf);
+ }
+
+ /**
+ * Creates a new instance of the <code>BytesWritable</code> object for the key
+ */
+ public Writable createValue() {
+ return (Writable)ReflectionUtils.newInstance(BytesWritable.class, conf);
+ }
+
+ /**
+ * Returns the current position in the file.
+ *
+ * @return The long of the current position in the file.
+ */
+ public long getPos()
+ throws IOException {
+ return in.getPos();
+ }
+
+ /**
+ * Returns the percentage of progress in processing the file. This will be
+ * represented as a float from 0 to 1 with 1 being 100% completed.
+ *
+ * @return The percentage of progress as a float from 0 to 1.
+ */
+ public float getProgress()
+ throws IOException {
+
+ // if we haven't even started
+ if (splitEnd == splitStart) {
+ return 0.0f;
+ }
+ else {
+ // the progress is current pos - where we started / length of the split
+ return Math.min(1.0f, (getPos() - splitStart) / (float)splitLen);
+ }
+ }
+
+ /**
+ * <p>Returns true if the next record in the split is read into the key and
+ * value pair. The key will be the arc record header and the values will be
+ * the raw content bytes of the arc record.</p>
+ *
+ * @param key The record key
+ * @param value The record value
+ *
+ * @return True if the next record is read.
+ *
+ * @throws IOException If an error occurs while reading the record value.
+ */
+ public boolean next(WritableComparable key, Writable value)
+ throws IOException {
+
+ try {
+
+ // get the starting position on the input stream
+ long startRead = in.getPos();
+ byte[] magicBuffer = null;
+
+ // we need this loop to handle false positives in reading of gzip records
+ while (true) {
+
+ // while we haven't passed the end of the split
+ if (startRead >= splitEnd) {
+ return false;
+ }
+
+ // scanning for the gzip header
+ boolean foundStart = false;
+ while (!foundStart) {
+
+ // start at the current file position and scan for 1K at time, break
+ // if there is no more to read
+ startRead = in.getPos();
+ magicBuffer = new byte[1024];
+ int read = in.read(magicBuffer);
+ if (read < 0) {
+ break;
+ }
+
+ // scan the byte array for the gzip header magic number. This happens
+ // byte by byte
+ for (int i = 0; i < read - 1; i++) {
+ byte[] testMagic = new byte[2];
+ System.arraycopy(magicBuffer, i, testMagic, 0, 2);
+ if (isMagic(testMagic)) {
+ // set the next start to the current gzip header
+ startRead += i;
+ foundStart = true;
+ break;
+ }
+ }
+ }
+
+ // seek to the start of the gzip header
+ in.seek(startRead);
+ ByteArrayOutputStream baos = null;
+ int totalRead = 0;
+
+ try {
+
+ // read 4K of the gzip at a time putting into a byte array
+ byte[] buffer = new byte[4096];
+ GZIPInputStream zin = new GZIPInputStream(in);
+ int gzipRead = -1;
+ baos = new ByteArrayOutputStream();
+ while ((gzipRead = zin.read(buffer, 0, buffer.length)) != -1) {
+ baos.write(buffer, 0, gzipRead);
+ totalRead += gzipRead;
+ }
+ }
+ catch (Exception e) {
+
+ // there are times we get false positives where the gzip header exists
+ // but it is not an actual gzip record, so we ignore it and start
+ // over seeking
+ System.out.println("Ignoring position: " + (startRead));
+ if (startRead + 1 < fileLen) {
+ in.seek(startRead + 1);
+ }
+ continue;
+ }
+
+ // change the output stream to a byte array
+ byte[] content = baos.toByteArray();
+
+ // the first line of the raw content in arc files is the header
+ int eol = 0;
+ for (int i = 0; i < content.length; i++) {
+ if (i > 0 && content[i] == '\n') {
+ eol = i;
+ break;
+ }
+ }
+
+ // create the header and the raw content minus the header
+ String header = new String(content, 0, eol).trim();
+ byte[] raw = new byte[(content.length - eol) - 1];
+ System.arraycopy(content, eol + 1, raw, 0, raw.length);
+
+ // populate key and values with the header and raw content.
+ Text keyText = (Text)key;
+ keyText.set(header);
+ BytesWritable valueBytes = (BytesWritable)value;
+ valueBytes.set(raw, 0, raw.length);
+
+ // TODO: It would be best to start at the end of the gzip read but
+ // the bytes read in gzip don't match raw bytes in the file so we
+ // overshoot the next header. With this current method you get
+ // some false positives but don't miss records.
+ if (startRead + 1 < fileLen) {
+ in.seek(startRead + 1);
+ }
+
+ // populated the record, now return
+ return true;
+ }
+ }
+ catch (Exception e) {
+ LOG.equals(StringUtils.stringifyException(e));
+ }
+
+ // couldn't populate the record or there is no next record to read
+ return false;
+ }
+}
Added: lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcSegmentCreator.java
URL: http://svn.apache.org/viewvc/lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcSegmentCreator.java?rev=591793&view=auto
==============================================================================
--- lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcSegmentCreator.java (added)
+++ lucene/nutch/trunk/src/java/org/apache/nutch/tools/arc/ArcSegmentCreator.java Sun Nov 4 08:01:53 2007
@@ -0,0 +1,402 @@
+/**
+ * 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.tools.arc;
+
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Map.Entry;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.util.ToolBase;
+import org.apache.nutch.crawl.CrawlDatum;
+import org.apache.nutch.crawl.NutchWritable;
+import org.apache.nutch.crawl.SignatureFactory;
+import org.apache.nutch.fetcher.FetcherOutputFormat;
+import org.apache.nutch.metadata.Metadata;
+import org.apache.nutch.metadata.Nutch;
+import org.apache.nutch.net.URLFilters;
+import org.apache.nutch.net.URLNormalizers;
+import org.apache.nutch.parse.Parse;
+import org.apache.nutch.parse.ParseImpl;
+import org.apache.nutch.parse.ParseResult;
+import org.apache.nutch.parse.ParseStatus;
+import org.apache.nutch.parse.ParseText;
+import org.apache.nutch.parse.ParseUtil;
+import org.apache.nutch.protocol.Content;
+import org.apache.nutch.protocol.ProtocolStatus;
+import org.apache.nutch.scoring.ScoringFilters;
+import org.apache.nutch.util.LogUtil;
+import org.apache.nutch.util.NutchConfiguration;
+import org.apache.nutch.util.NutchJob;
+import org.apache.nutch.util.StringUtil;
+
+/**
+ * <p>The <code>ArcSegmentCreator</code> is a replacement for fetcher that will
+ * take arc files as input and produce a nutch segment as output.</p>
+ *
+ * <p>Arc files are tars of compressed gzips which are produced by both the
+ * internet archive project and the grub distributed crawler project.</p>
+ *
+ * TODO: This class needs to be changed to use ToolRunner instead of ToolBase.
+ */
+public class ArcSegmentCreator
+ extends ToolBase
+ implements Mapper {
+
+ public static final Log LOG = LogFactory.getLog(ArcSegmentCreator.class);
+ public static final String URL_VERSION = "arc.url.version";
+ private JobConf jobConf;
+ private URLFilters urlFilters;
+ private ScoringFilters scfilters;
+ private ParseUtil parseUtil;
+ private URLNormalizers normalizers;
+ private int interval;
+
+ private static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
+
+ public ArcSegmentCreator() {
+
+ }
+
+ /**
+ * <p>Constructor that sets the job configuration.</p>
+ *
+ * @param conf
+ */
+ public ArcSegmentCreator(Configuration conf) {
+ setConf(conf);
+ }
+
+ /**
+ * Generates a random name for the segments.
+ *
+ * @return The generated segment name.
+ */
+ public static synchronized String generateSegmentName() {
+ try {
+ Thread.sleep(1000);
+ }
+ catch (Throwable t) {
+ }
+ return sdf.format(new Date(System.currentTimeMillis()));
+ }
+
+ /**
+ * <p>Configures the job. Sets the url filters, scoring filters, url normalizers
+ * and other relevant data.</p>
+ *
+ * @param job The job configuration.
+ */
+ public void configure(JobConf job) {
+
+ // set the url filters, scoring filters the parse util and the url
+ // normalizers
+ this.jobConf = job;
+ this.urlFilters = new URLFilters(jobConf);
+ this.scfilters = new ScoringFilters(jobConf);
+ this.parseUtil = new ParseUtil(jobConf);
+ this.normalizers = new URLNormalizers(jobConf, URLNormalizers.SCOPE_FETCHER);
+ interval = jobConf.getInt("db.fetch.interval.default", 2592000);
+ }
+
+ public void close() {
+ }
+
+ /**
+ * <p>Parses the raw content of a single record to create output. This method
+ * is almost the same as the {@link org.apache.nutch.Fetcher#output} method in
+ * terms of processing and output.
+ *
+ * @param output The job output collector.
+ * @param segmentName The name of the segment to create.
+ * @param key The url of the record.
+ * @param datum The CrawlDatum of the record.
+ * @param content The raw content of the record
+ * @param pstatus The protocol status
+ * @param status The fetch status.
+ *
+ * @return The result of the parse in a ParseStatus object.
+ */
+ private ParseStatus output(OutputCollector output, String segmentName,
+ Text key, CrawlDatum datum, Content content, ProtocolStatus pstatus,
+ int status) {
+
+ // set the fetch status and the fetch time
+ datum.setStatus(status);
+ datum.setFetchTime(System.currentTimeMillis());
+ if (pstatus != null)
+ datum.getMetaData().put(Nutch.WRITABLE_PROTO_STATUS_KEY, pstatus);
+
+ ParseResult parseResult = null;
+ if (content != null) {
+ Metadata metadata = content.getMetadata();
+ // add segment to metadata
+ metadata.set(Nutch.SEGMENT_NAME_KEY, segmentName);
+ // add score to content metadata so that ParseSegment can pick it up.
+ try {
+ scfilters.passScoreBeforeParsing(key, datum, content);
+ }
+ catch (Exception e) {
+ if (LOG.isWarnEnabled()) {
+ e.printStackTrace(LogUtil.getWarnStream(LOG));
+ LOG.warn("Couldn't pass score, url " + key + " (" + e + ")");
+ }
+ }
+
+ try {
+
+ // parse the content
+ parseResult = this.parseUtil.parse(content);
+ }
+ catch (Exception e) {
+ LOG.warn("Error parsing: " + key + ": "
+ + StringUtils.stringifyException(e));
+ }
+
+ // set the content signature
+ if (parseResult == null) {
+ byte[] signature = SignatureFactory.getSignature(getConf()).calculate(
+ content, new ParseStatus().getEmptyParse(conf));
+ datum.setSignature(signature);
+ }
+
+ try {
+ output.collect(key, new NutchWritable(datum));
+ output.collect(key, new NutchWritable(content));
+
+ if (parseResult != null) {
+ for (Entry <Text, Parse> entry : parseResult) {
+ Text url = entry.getKey();
+ Parse parse = entry.getValue();
+ ParseStatus parseStatus = parse.getData().getStatus();
+
+ if (!parseStatus.isSuccess()) {
+ LOG.warn("Error parsing: " + key + ": " + parseStatus);
+ parse = parseStatus.getEmptyParse(getConf());
+ }
+
+ // Calculate page signature.
+ byte[] signature = SignatureFactory.getSignature(getConf()).calculate(
+ content, parse);
+ // Ensure segment name and score are in parseData metadata
+ parse.getData().getContentMeta().set(Nutch.SEGMENT_NAME_KEY,
+ segmentName);
+ parse.getData().getContentMeta().set(Nutch.SIGNATURE_KEY,
+ StringUtil.toHexString(signature));
+ // Pass fetch time to content meta
+ parse.getData().getContentMeta().set(Nutch.FETCH_TIME_KEY,
+ Long.toString(datum.getFetchTime()));
+ if (url.equals(key))
+ datum.setSignature(signature);
+ try {
+ scfilters.passScoreAfterParsing(url, content, parse);
+ }
+ catch (Exception e) {
+ if (LOG.isWarnEnabled()) {
+ e.printStackTrace(LogUtil.getWarnStream(LOG));
+ LOG.warn("Couldn't pass score, url " + key + " (" + e + ")");
+ }
+ }
+ output.collect(url, new NutchWritable(new ParseImpl(new ParseText(
+ parse.getText()), parse.getData(), parse.isCanonical())));
+ }
+ }
+ }
+ catch (IOException e) {
+ if (LOG.isFatalEnabled()) {
+ LOG.fatal("ArcSegmentCreator caught:" + StringUtils.stringifyException(e));
+ }
+ }
+
+ // return parse status if it exits
+ if (parseResult != null && !parseResult.isEmpty()) {
+ Parse p = parseResult.get(content.getUrl());
+ if (p != null) {
+ return p.getData().getStatus();
+ }
+ }
+ }
+
+ return null;
+ }
+
+ /**
+ * <p>Logs any error that occurs during conversion.</p>
+ *
+ * @param url The url we are parsing.
+ * @param t The error that occured.
+ */
+ private void logError(Text url, Throwable t) {
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Conversion of " + url + " failed with: " +
+ StringUtils.stringifyException(t));
+ }
+ }
+
+ /**
+ * <p>Runs the Map job to translate an arc record into output for Nutch
+ * segments.</p>
+ *
+ * @param key The arc record header.
+ * @param value The arc record raw content bytes.
+ * @param output The output collecter.
+ * @param reporter The progress reporter.
+ */
+ public void map(WritableComparable key, Writable value,
+ OutputCollector output, Reporter reporter)
+ throws IOException {
+
+ String[] headers = key.toString().split("\\s+");
+ String urlStr = headers[0];
+ String version = headers[2];
+ String contentType = headers[3];
+
+ // arcs start with a file description. for now we ignore this as it is not
+ // a content record
+ if (urlStr.startsWith("filedesc://")) {
+ LOG.info("Ignoring file header: " + urlStr);
+ return;
+ }
+ LOG.info("Processing: " + urlStr);
+
+ // get the raw bytes from the arc file, create a new crawldatum
+ Text url = new Text();
+ BytesWritable bytes = (BytesWritable)value;
+ CrawlDatum datum = new CrawlDatum(CrawlDatum.STATUS_DB_FETCHED, interval,
+ 1.0f);
+ String segmentName = getConf().get(Nutch.SEGMENT_NAME_KEY);
+
+ // normalize and filter the urls
+ try {
+ urlStr = normalizers.normalize(urlStr, URLNormalizers.SCOPE_FETCHER);
+ urlStr = urlFilters.filter(urlStr); // filter the url
+ }
+ catch (Exception e) {
+ if (LOG.isWarnEnabled()) {
+ LOG.warn("Skipping " + url + ":" + e);
+ }
+ urlStr = null;
+ }
+
+ // if still a good url then process
+ if (urlStr != null) {
+
+ url.set(urlStr);
+ try {
+
+ // set the protocol status to success and the crawl status to success
+ // create the content from the normalized url and the raw bytes from
+ // the arc file, TODO: currently this doesn't handle text of errors
+ // pages (i.e. 404, etc.). We assume we won't get those.
+ ProtocolStatus status = ProtocolStatus.STATUS_SUCCESS;
+ Content content = new Content(urlStr, urlStr, bytes.get(), contentType,
+ new Metadata(), getConf());
+
+ // set the url version into the metadata
+ content.getMetadata().set(URL_VERSION, version);
+ ParseStatus pstatus = null;
+ pstatus = output(output, segmentName, url, datum, content, status,
+ CrawlDatum.STATUS_FETCH_SUCCESS);
+ reporter.progress();
+ }
+ catch (Throwable t) { // unexpected exception
+ logError(url, t);
+ output(output, segmentName, url, datum, null, null,
+ CrawlDatum.STATUS_FETCH_RETRY);
+ }
+ }
+ }
+
+ /**
+ * <p>Creates the arc files to segments job.</p>
+ *
+ * @param arcFiles The path to the directory holding the arc files
+ * @param segmentsOutDir The output directory for writing the segments
+ *
+ * @throws IOException If an IO error occurs while running the job.
+ */
+ public void createSegments(Path arcFiles, Path segmentsOutDir)
+ throws IOException {
+
+ if (LOG.isInfoEnabled()) {
+ LOG.info("ArcSegmentCreator: starting");
+ LOG.info("ArcSegmentCreator: arc files dir: " + arcFiles);
+ }
+
+ JobConf job = new NutchJob(getConf());
+ job.setJobName("ArcSegmentCreator " + arcFiles);
+ String segName = generateSegmentName();
+ job.set(Nutch.SEGMENT_NAME_KEY, segName);
+ job.setInputPath(arcFiles);
+ job.setInputFormat(ArcInputFormat.class);
+ job.setMapperClass(ArcSegmentCreator.class);
+ job.setOutputPath(new Path(segmentsOutDir, segName));
+ job.setOutputFormat(FetcherOutputFormat.class);
+ job.setOutputKeyClass(Text.class);
+ job.setOutputValueClass(NutchWritable.class);
+
+ JobClient.runJob(job);
+ if (LOG.isInfoEnabled()) {
+ LOG.info("ArcSegmentCreator: done");
+ }
+ }
+
+ public static void main(String args[])
+ throws Exception {
+ int res = new ArcSegmentCreator().doMain(NutchConfiguration.create(), args);
+ System.exit(res);
+ }
+
+ public int run(String[] args)
+ throws Exception {
+
+ String usage = "Usage: ArcSegmentCreator <arcFiles> <segmentsOutDir>";
+
+ if (args.length < 2) {
+ System.err.println(usage);
+ return -1;
+ }
+
+ // set the arc files directory and the segments output directory
+ Path arcFiles = new Path(args[0]);
+ Path segmentsOutDir = new Path(args[1]);
+
+ try {
+ // create the segments from the arc files
+ createSegments(arcFiles, segmentsOutDir);
+ return 0;
+ }
+ catch (Exception e) {
+ LOG.fatal("ArcSegmentCreator: " + StringUtils.stringifyException(e));
+ return -1;
+ }
+ }
+}