You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nutch.apache.org by do...@apache.org on 2009/08/17 00:25:17 UTC
svn commit: r804789 [4/6] - in /lucene/nutch/branches/nutchbase: ./ bin/
conf/ lib/ src/java/org/apache/nutch/analysis/
src/java/org/apache/nutch/crawl/ src/java/org/apache/nutch/fetcher/
src/java/org/apache/nutch/indexer/ src/java/org/apache/nutch/ind...
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/IndexingFilter.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/IndexingFilter.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/IndexingFilter.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/IndexingFilter.java Sun Aug 16 22:25:12 2009
@@ -20,20 +20,19 @@
// Hadoop imports
import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.Text;
// Nutch imports
-import org.apache.nutch.parse.Parse;
-import org.apache.nutch.crawl.CrawlDatum;
-import org.apache.nutch.crawl.Inlinks;
-import org.apache.nutch.plugin.Pluggable;
+import org.apache.nutch.indexer.IndexingException;
+import org.apache.nutch.indexer.NutchDocument;
+import org.apache.nutch.plugin.TablePluggable;
+import org.apache.nutch.util.hbase.WebTableRow;
/** Extension point for indexing. Permits one to add metadata to the indexed
* fields. All plugins found which implement this extension point are run
* sequentially on the parse.
*/
-public interface IndexingFilter extends Pluggable, Configurable {
+public interface IndexingFilter extends TablePluggable, Configurable {
/** The name of the extension point. */
final static String X_POINT_ID = IndexingFilter.class.getName();
@@ -42,17 +41,15 @@
* parse. Unwanted documents can be removed from indexing by returning a null value.
*
* @param doc document instance for collecting fields
- * @param parse parse data instance
* @param url page url
- * @param datum crawl datum for the page
- * @param inlinks page inlinks
+ * @param url's row
* @return modified (or a new) document instance, or null (meaning the document
* should be discarded)
* @throws IndexingException
*/
- NutchDocument filter(NutchDocument doc, Parse parse, Text url, CrawlDatum datum, Inlinks inlinks)
- throws IndexingException;
-
+ NutchDocument filter(NutchDocument doc, String url, WebTableRow row)
+ throws IndexingException;
+
/** Adds index-level configuraition options.
* Implementations can update given configuration to pass document-independent
* information to indexing backends. As a rule of thumb, prefix meta keys
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/IndexingFilters.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/IndexingFilters.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/IndexingFilters.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/IndexingFilters.java Sun Aug 16 22:25:12 2009
@@ -18,24 +18,26 @@
package org.apache.nutch.indexer;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
+import java.util.HashSet;
// Commons Logging imports
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.nutch.plugin.*;
-import org.apache.nutch.parse.Parse;
import org.apache.nutch.util.ObjectCache;
+import org.apache.nutch.util.hbase.HbaseColumn;
+import org.apache.nutch.util.hbase.WebTableRow;
import org.apache.hadoop.conf.Configuration;
-import org.apache.nutch.crawl.CrawlDatum;
-import org.apache.nutch.crawl.Inlinks;
-import org.apache.hadoop.io.Text;
+import org.apache.nutch.indexer.IndexingException;
+import org.apache.nutch.indexer.NutchDocument;
/** Creates and caches {@link IndexingFilter} implementing plugins.*/
public class IndexingFilters {
- public static final String INDEXINGFILTER_ORDER = "indexingfilter.order";
+ public static final String INDEXINGFILTER_ORDER = "indexingfilterhbase.order";
public final static Log LOG = LogFactory.getLog(IndexingFilters.class);
@@ -86,8 +88,7 @@
} else {
ArrayList<IndexingFilter> filters = new ArrayList<IndexingFilter>();
for (int i = 0; i < orderedFilters.length; i++) {
- IndexingFilter filter = filterMap
- .get(orderedFilters[i]);
+ IndexingFilter filter = filterMap.get(orderedFilters[i]);
if (filter != null) {
filter.addIndexBackendOptions(conf);
filters.add(filter);
@@ -102,13 +103,12 @@
this.indexingFilters = (IndexingFilter[]) objectCache
.getObject(IndexingFilter.class.getName());
}
- }
-
+ }
/** Run all defined filters. */
- public NutchDocument filter(NutchDocument doc, Parse parse, Text url, CrawlDatum datum,
- Inlinks inlinks) throws IndexingException {
- for (int i = 0; i < this.indexingFilters.length; i++) {
- doc = this.indexingFilters[i].filter(doc, parse, url, datum, inlinks);
+ public NutchDocument filter(NutchDocument doc, String url, WebTableRow row)
+ throws IndexingException {
+ for (IndexingFilter indexingFilter : indexingFilters) {
+ doc = indexingFilter.filter(doc, url, row);
// break the loop if an indexing filter discards the doc
if (doc == null) return null;
}
@@ -116,4 +116,12 @@
return doc;
}
+ public Collection<HbaseColumn> getColumns() {
+ Collection<HbaseColumn> columns = new HashSet<HbaseColumn>();
+ for (IndexingFilter indexingFilter : indexingFilters) {
+ columns.addAll(indexingFilter.getColumns());
+ }
+ return columns;
+ }
+
}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/NutchIndexWriter.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/NutchIndexWriter.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/NutchIndexWriter.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/NutchIndexWriter.java Sun Aug 16 22:25:12 2009
@@ -18,10 +18,11 @@
import java.io.IOException;
-import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.nutch.indexer.NutchDocument;
public interface NutchIndexWriter {
- public void open(JobConf job, String name) throws IOException;
+ public void open(TaskAttemptContext job, String name) throws IOException;
public void write(NutchDocument doc) throws IOException;
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/lucene/LuceneWriter.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/lucene/LuceneWriter.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/lucene/LuceneWriter.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/lucene/LuceneWriter.java Sun Aug 16 22:25:12 2009
@@ -27,11 +27,12 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapred.FileOutputFormat;
-import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriter.MaxFieldLength;
import org.apache.nutch.analysis.AnalyzerFactory;
import org.apache.nutch.analysis.NutchAnalyzer;
import org.apache.nutch.analysis.NutchDocumentAnalyzer;
@@ -39,6 +40,7 @@
import org.apache.nutch.indexer.NutchDocument;
import org.apache.nutch.indexer.NutchIndexWriter;
import org.apache.nutch.indexer.NutchSimilarity;
+import org.apache.nutch.indexer.lucene.LuceneConstants;
import org.apache.nutch.metadata.Metadata;
import org.apache.nutch.util.LogUtil;
@@ -108,13 +110,13 @@
} else if (LuceneConstants.STORE_NO.equals(val)) {
store = Field.Store.NO;
} else if (LuceneConstants.INDEX_TOKENIZED.equals(val)) {
- index = Field.Index.TOKENIZED;
+ index = Field.Index.ANALYZED;
} else if (LuceneConstants.INDEX_NO.equals(val)) {
index = Field.Index.NO;
} else if (LuceneConstants.INDEX_UNTOKENIZED.equals(val)) {
- index = Field.Index.UN_TOKENIZED;
+ index = Field.Index.NOT_ANALYZED;
} else if (LuceneConstants.INDEX_NO_NORMS.equals(val)) {
- index = Field.Index.NO_NORMS;
+ index = Field.Index.ANALYZED_NO_NORMS;
} else if (LuceneConstants.VECTOR_NO.equals(val)) {
vector = Field.TermVector.NO;
} else if (LuceneConstants.VECTOR_YES.equals(val)) {
@@ -169,13 +171,13 @@
fieldIndex.put(field, Field.Index.NO);
break;
case NO_NORMS:
- fieldIndex.put(field, Field.Index.NO_NORMS);
+ fieldIndex.put(field, Field.Index.ANALYZED_NO_NORMS);
break;
case TOKENIZED:
- fieldIndex.put(field, Field.Index.TOKENIZED);
+ fieldIndex.put(field, Field.Index.ANALYZED);
break;
case UNTOKENIZED:
- fieldIndex.put(field, Field.Index.UN_TOKENIZED);
+ fieldIndex.put(field, Field.Index.NOT_ANALYZED);
break;
}
} else if (key.startsWith(LuceneConstants.FIELD_VECTOR_PREFIX)) {
@@ -203,32 +205,35 @@
}
}
- public void open(JobConf job, String name)
+ public void open(TaskAttemptContext job, String name)
throws IOException {
- this.fs = FileSystem.get(job);
+ Configuration conf = job.getConfiguration();
+ this.fs = FileSystem.get(conf);
perm = new Path(FileOutputFormat.getOutputPath(job), name);
- temp = job.getLocalPath("index/_" +
+ temp = conf.getLocalPath("mapred.local.dir", "index/_" +
Integer.toString(new Random().nextInt()));
fs.delete(perm, true); // delete old, if any
- analyzerFactory = new AnalyzerFactory(job);
+ analyzerFactory = new AnalyzerFactory(conf);
+ MaxFieldLength mfl = new MaxFieldLength(IndexWriter.DEFAULT_MAX_FIELD_LENGTH);
writer = new IndexWriter(fs.startLocalOutput(perm, temp).toString(),
- new NutchDocumentAnalyzer(job), true);
+ new NutchDocumentAnalyzer(conf), true, mfl);
- writer.setMergeFactor(job.getInt("indexer.mergeFactor", 10));
- writer.setMaxBufferedDocs(job.getInt("indexer.minMergeDocs", 100));
- writer.setMaxMergeDocs(job
+ writer.setMergeFactor(conf.getInt("indexer.mergeFactor", 10));
+ writer.setMaxBufferedDocs(conf.getInt("indexer.minMergeDocs", 100));
+ writer.setMaxMergeDocs(conf
.getInt("indexer.maxMergeDocs", Integer.MAX_VALUE));
- writer.setTermIndexInterval(job.getInt("indexer.termIndexInterval", 128));
- writer.setMaxFieldLength(job.getInt("indexer.max.tokens", 10000));
+ writer.setTermIndexInterval(conf.getInt("indexer.termIndexInterval", 128));
+ writer.setMaxFieldLength(conf.getInt("indexer.max.tokens", 10000));
writer.setInfoStream(LogUtil.getDebugStream(Indexer.LOG));
writer.setUseCompoundFile(false);
writer.setSimilarity(new NutchSimilarity());
- processOptions(job);
+ processOptions(conf);
}
public void close() throws IOException {
+ writer.commit();
writer.optimize();
writer.close();
fs.completeLocalOutput(perm, temp); // copy to dfs
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrDeleteDuplicates.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrDeleteDuplicates.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrDeleteDuplicates.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrDeleteDuplicates.java Sun Aug 16 22:25:12 2009
@@ -4,28 +4,28 @@
import java.io.DataOutput;
import java.io.IOException;
import java.net.MalformedURLException;
+import java.util.ArrayList;
import java.util.Iterator;
+import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.mapred.InputFormat;
-import org.apache.hadoop.mapred.InputSplit;
-import org.apache.hadoop.mapred.JobClient;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.OutputCollector;
-import org.apache.hadoop.mapred.RecordReader;
-import org.apache.hadoop.mapred.Reducer;
-import org.apache.hadoop.mapred.Reporter;
-import org.apache.hadoop.mapred.lib.IdentityMapper;
-import org.apache.hadoop.mapred.lib.NullOutputFormat;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.nutch.indexer.DeleteDuplicates;
import org.apache.nutch.util.NutchConfiguration;
-import org.apache.nutch.util.NutchJob;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServer;
import org.apache.solr.client.solrj.SolrServerException;
@@ -65,8 +65,8 @@
* documents with <b>different</b> URLs but the same digest.
*/
public class SolrDeleteDuplicates
-implements Reducer<Text, SolrDeleteDuplicates.SolrRecord, Text, SolrDeleteDuplicates.SolrRecord>,
-Tool {
+extends Reducer<Text, SolrDeleteDuplicates.SolrRecord, Text, SolrDeleteDuplicates.SolrRecord>
+implements Tool {
public static final Log LOG = LogFactory.getLog(SolrDeleteDuplicates.class);
@@ -121,7 +121,7 @@
}
}
- public static class SolrInputSplit implements InputSplit {
+ public static class SolrInputSplit extends InputSplit {
private int docBegin;
private int numDocs;
@@ -137,10 +137,6 @@
return docBegin;
}
- public int getNumDocs() {
- return numDocs;
- }
-
@Override
public long getLength() throws IOException {
return numDocs;
@@ -150,25 +146,72 @@
public String[] getLocations() throws IOException {
return new String[] {} ;
}
+ }
+
+ public static class SolrRecordReader extends RecordReader<Text, SolrRecord> {
+ private int currentDoc = 0;
+ private int numDocs;
+ private Text text;
+ private SolrRecord record;
+ private SolrDocumentList solrDocs;
+
+ public SolrRecordReader(SolrDocumentList solrDocs, int numDocs) {
+ this.solrDocs = solrDocs;
+ this.numDocs = numDocs;
+ }
+
@Override
- public void readFields(DataInput in) throws IOException {
- docBegin = in.readInt();
- numDocs = in.readInt();
+ public void initialize(InputSplit split, TaskAttemptContext context)
+ throws IOException, InterruptedException {
+ text = new Text();
+ record = new SolrRecord();
}
@Override
- public void write(DataOutput out) throws IOException {
- out.writeInt(docBegin);
- out.writeInt(numDocs);
+ public void close() throws IOException { }
+
+ @Override
+ public float getProgress() throws IOException {
+ return currentDoc / (float) numDocs;
}
- }
- public static class SolrInputFormat implements InputFormat<Text, SolrRecord> {
+ @Override
+ public Text getCurrentKey() throws IOException, InterruptedException {
+ return text;
+ }
+
+ @Override
+ public SolrRecord getCurrentValue() throws IOException,
+ InterruptedException {
+ return record;
+ }
+
+ @Override
+ public boolean nextKeyValue() throws IOException, InterruptedException {
+ if (currentDoc >= numDocs) {
+ return false;
+ }
+
+ SolrDocument doc = solrDocs.get(currentDoc);
+ String digest = (String) doc.getFieldValue(SolrConstants.DIGEST_FIELD);
+ text.set(digest);
+ record.readSolrDocument(doc);
- /** Return each index as a split. */
- public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
- SolrServer solr = new CommonsHttpSolrServer(job.get(SolrConstants.SERVER_URL));
+ currentDoc++;
+ return true;
+ }
+
+ };
+
+ public static class SolrInputFormat extends InputFormat<Text, SolrRecord> {
+
+ @Override
+ public List<InputSplit> getSplits(JobContext context)
+ throws IOException, InterruptedException {
+ Configuration conf = context.getConfiguration();
+ int numSplits = context.getNumReduceTasks();
+ SolrServer solr = new CommonsHttpSolrServer(conf.get(SolrConstants.SERVER_URL));
final SolrQuery solrQuery = new SolrQuery(SOLR_GET_ALL_QUERY);
solrQuery.setFields(SolrConstants.ID_FIELD);
@@ -184,24 +227,23 @@
int numResults = (int)response.getResults().getNumFound();
int numDocsPerSplit = (numResults / numSplits);
int currentDoc = 0;
- SolrInputSplit[] splits = new SolrInputSplit[numSplits];
+ List<InputSplit> splits = new ArrayList<InputSplit>();
for (int i = 0; i < numSplits - 1; i++) {
- splits[i] = new SolrInputSplit(currentDoc, numDocsPerSplit);
+ splits.add(new SolrInputSplit(currentDoc, numDocsPerSplit));
currentDoc += numDocsPerSplit;
}
- splits[splits.length - 1] = new SolrInputSplit(currentDoc, numResults - currentDoc);
+ splits.add(new SolrInputSplit(currentDoc, numResults - currentDoc));
return splits;
}
- public RecordReader<Text, SolrRecord> getRecordReader(final InputSplit split,
- final JobConf job,
- Reporter reporter)
- throws IOException {
-
- SolrServer solr = new CommonsHttpSolrServer(job.get(SolrConstants.SERVER_URL));
+ @Override
+ public RecordReader<Text, SolrRecord> createRecordReader(InputSplit split,
+ TaskAttemptContext context) throws IOException, InterruptedException {
+ Configuration conf = context.getConfiguration();
+ SolrServer solr = new CommonsHttpSolrServer(conf.get(SolrConstants.SERVER_URL));
SolrInputSplit solrSplit = (SolrInputSplit) split;
- final int numDocs = solrSplit.getNumDocs();
+ final int numDocs = (int) solrSplit.getLength();
SolrQuery solrQuery = new SolrQuery(SOLR_GET_ALL_QUERY);
solrQuery.setFields(SolrConstants.ID_FIELD, SolrConstants.BOOST_FIELD,
@@ -218,49 +260,7 @@
}
final SolrDocumentList solrDocs = response.getResults();
-
- return new RecordReader<Text, SolrRecord>() {
-
- private int currentDoc = 0;
-
- @Override
- public void close() throws IOException { }
-
- @Override
- public Text createKey() {
- return new Text();
- }
-
- @Override
- public SolrRecord createValue() {
- return new SolrRecord();
- }
-
- @Override
- public long getPos() throws IOException {
- return currentDoc;
- }
-
- @Override
- public float getProgress() throws IOException {
- return currentDoc / (float) numDocs;
- }
-
- @Override
- public boolean next(Text key, SolrRecord value) throws IOException {
- if (currentDoc >= numDocs) {
- return false;
- }
-
- SolrDocument doc = solrDocs.get(currentDoc);
- String digest = (String) doc.getFieldValue(SolrConstants.DIGEST_FIELD);
- key.set(digest);
- value.readSolrDocument(doc);
-
- currentDoc++;
- return true;
- }
- };
+ return new SolrRecordReader(solrDocs, numDocs);
}
}
@@ -283,34 +283,34 @@
}
@Override
- public void configure(JobConf job) {
+ public void setup(Context job) throws IOException {
+ Configuration conf = job.getConfiguration();
try {
- solr = new CommonsHttpSolrServer(job.get(SolrConstants.SERVER_URL));
+ solr = new CommonsHttpSolrServer(conf.get(SolrConstants.SERVER_URL));
} catch (MalformedURLException e) {
- throw new RuntimeException(e);
+ throw new IOException(e);
}
}
@Override
- public void close() throws IOException {
+ public void cleanup(Context context) throws IOException {
try {
if (numDeletes > 0) {
updateRequest.process(solr);
}
- solr.optimize();
} catch (SolrServerException e) {
throw new IOException(e);
}
}
@Override
- public void reduce(Text key, Iterator<SolrRecord> values,
- OutputCollector<Text, SolrRecord> output, Reporter reporter)
+ public void reduce(Text key, Iterable<SolrRecord> values, Context context)
throws IOException {
- SolrRecord recordToKeep = values.next();
- while (values.hasNext()) {
- SolrRecord solrRecord = values.next();
+ Iterator<SolrRecord> iterator = values.iterator();
+ SolrRecord recordToKeep = iterator.next();
+ while (iterator.hasNext()) {
+ SolrRecord solrRecord = iterator.next();
if (solrRecord.getBoost() > recordToKeep.getBoost() ||
(solrRecord.getBoost() == recordToKeep.getBoost() &&
solrRecord.getTstamp() > recordToKeep.getTstamp())) {
@@ -332,33 +332,39 @@
}
}
- public void dedup(String solrUrl) throws IOException {
+ public boolean dedup(String solrUrl)
+ throws IOException, InterruptedException, ClassNotFoundException {
LOG.info("SolrDeleteDuplicates: starting...");
LOG.info("SolrDeleteDuplicates: Solr url: " + solrUrl);
- JobConf job = new NutchJob(getConf());
+ getConf().set(SolrConstants.SERVER_URL, solrUrl);
+
+ Job job = new Job(getConf(), "solrdedup");
- job.set(SolrConstants.SERVER_URL, solrUrl);
- job.setInputFormat(SolrInputFormat.class);
- job.setOutputFormat(NullOutputFormat.class);
+ job.setInputFormatClass(SolrInputFormat.class);
+ job.setOutputFormatClass(NullOutputFormat.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(SolrRecord.class);
- job.setMapperClass(IdentityMapper.class);
+ job.setMapperClass(Mapper.class);
job.setReducerClass(SolrDeleteDuplicates.class);
- JobClient.runJob(job);
-
- LOG.info("SolrDeleteDuplicates: done.");
+ return job.waitForCompletion(true);
}
- public int run(String[] args) throws IOException {
+ public int run(String[] args)
+ throws IOException, InterruptedException, ClassNotFoundException {
if (args.length != 1) {
System.err.println("Usage: SolrDeleteDuplicates <solr url>");
return 1;
}
- dedup(args[0]);
- return 0;
+ boolean result = dedup(args[0]);
+ if (result) {
+ LOG.info("SolrDeleteDuplicates: done.");
+ return 0;
+ }
+
+ return -1;
}
public static void main(String[] args) throws Exception {
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrIndexer.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrIndexer.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrIndexer.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrIndexer.java Sun Aug 16 22:25:12 2009
@@ -16,9 +16,6 @@
*/
package org.apache.nutch.indexer.solr;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
import java.util.Random;
import org.apache.commons.logging.Log;
@@ -27,16 +24,15 @@
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapred.FileOutputFormat;
-import org.apache.hadoop.mapred.JobClient;
-import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
-import org.apache.nutch.indexer.IndexerMapReduce;
+import org.apache.nutch.indexer.Indexer;
import org.apache.nutch.indexer.NutchIndexWriterFactory;
+import org.apache.nutch.indexer.solr.SolrConstants;
import org.apache.nutch.util.NutchConfiguration;
-import org.apache.nutch.util.NutchJob;
public class SolrIndexer extends Configured implements Tool {
@@ -50,49 +46,40 @@
super(conf);
}
- private void indexSolr(String solrUrl, Path crawlDb, Path linkDb,
- List<Path> segments) throws IOException {
+ private void indexSolr(String solrUrl, String table, boolean reindex)
+ throws Exception {
LOG.info("SolrIndexer: starting");
- final JobConf job = new NutchJob(getConf());
- job.setJobName("index-solr " + solrUrl);
-
- IndexerMapReduce.initMRJob(crawlDb, linkDb, segments, job);
-
- job.set(SolrConstants.SERVER_URL, solrUrl);
-
- NutchIndexWriterFactory.addClassToConf(job, SolrWriter.class);
-
- job.setReduceSpeculativeExecution(false);
+ NutchIndexWriterFactory.addClassToConf(getConf(), SolrWriter.class);
+ getConf().set(SolrConstants.SERVER_URL, solrUrl);
+ final Job job = Indexer.createIndexJob(getConf(), "solr-index " + table,
+ table, reindex);;
final Path tmp = new Path("tmp_" + System.currentTimeMillis() + "-" +
new Random().nextInt());
FileOutputFormat.setOutputPath(job, tmp);
try {
- JobClient.runJob(job);
+ job.waitForCompletion(true);
} finally {
- FileSystem.get(job).delete(tmp, true);
+ FileSystem.get(getConf()).delete(tmp, true);
}
LOG.info("SolrIndexer: done");
}
public int run(String[] args) throws Exception {
- if (args.length < 4) {
- System.err.println("Usage: SolrIndexer <solr url> <crawldb> <linkdb> <segment> ...");
+ if (args.length < 2) {
+ System.err.println("Usage: SolrIndexer <solr url> <webtable> [-reindex]");
return -1;
}
- final Path crawlDb = new Path(args[1]);
- final Path linkDb = new Path(args[2]);
-
- final List<Path> segments = new ArrayList<Path>();
- for (int i = 3; i < args.length; i++) {
- segments.add(new Path(args[i]));
+ boolean reindex = false;
+ if (args.length > 2 && "-reindex".equals(args[2])) {
+ reindex = true;
}
try {
- indexSolr(args[0], crawlDb, linkDb, segments);
+ indexSolr(args[0], args[1], reindex);
return 0;
} catch (final Exception e) {
LOG.fatal("SolrIndexer: " + StringUtils.stringifyException(e));
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrWriter.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrWriter.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrWriter.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/indexer/solr/SolrWriter.java Sun Aug 16 22:25:12 2009
@@ -21,7 +21,8 @@
import java.util.List;
import java.util.Map.Entry;
-import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.nutch.indexer.NutchDocument;
import org.apache.nutch.indexer.NutchIndexWriter;
import org.apache.solr.client.solrj.SolrServer;
@@ -38,12 +39,15 @@
private int commitSize;
- public void open(JobConf job, String name)
+ @Override
+ public void open(TaskAttemptContext job, String name)
throws IOException {
- solr = new CommonsHttpSolrServer(job.get(SolrConstants.SERVER_URL));
- commitSize = job.getInt(SolrConstants.COMMIT_SIZE, 1000);
+ Configuration conf = job.getConfiguration();
+ solr = new CommonsHttpSolrServer(conf.get(SolrConstants.SERVER_URL));
+ commitSize = conf.getInt(SolrConstants.COMMIT_SIZE, 1000);
}
+ @Override
public void write(NutchDocument doc) throws IOException {
final SolrInputDocument inputDoc = new SolrInputDocument();
for(final Entry<String, List<String>> e : doc) {
@@ -57,12 +61,13 @@
try {
solr.add(inputDocs);
} catch (final SolrServerException e) {
- throw makeIOException(e);
+ throw new IOException(e);
}
inputDocs.clear();
}
}
+ @Override
public void close() throws IOException {
try {
if (!inputDocs.isEmpty()) {
@@ -71,14 +76,8 @@
}
solr.commit();
} catch (final SolrServerException e) {
- throw makeIOException(e);
+ throw new IOException(e);
}
}
- public static IOException makeIOException(SolrServerException e) {
- final IOException ioe = new IOException();
- ioe.initCause(e);
- return ioe;
- }
-
}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HTMLMetaTags.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HTMLMetaTags.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HTMLMetaTags.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HTMLMetaTags.java Sun Aug 16 22:25:12 2009
@@ -188,7 +188,7 @@
+ ", refreshHref=" + refreshHref + "\n"
);
sb.append(" * general tags:\n");
- Iterator it = generalTags.keySet().iterator();
+ Iterator<?> it = generalTags.keySet().iterator();
while (it.hasNext()) {
String key = (String)it.next();
sb.append(" - " + key + "\t=\t" + generalTags.get(key) + "\n");
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HtmlParseFilter.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HtmlParseFilter.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HtmlParseFilter.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HtmlParseFilter.java Sun Aug 16 22:25:12 2009
@@ -1,42 +1,27 @@
-/**
- * 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.parse;
-// JDK imports
import org.w3c.dom.DocumentFragment;
// Hadoop imports
import org.apache.hadoop.conf.Configurable;
// Nutch imports
-import org.apache.nutch.plugin.Pluggable;
-import org.apache.nutch.protocol.Content;
+import org.apache.nutch.parse.HTMLMetaTags;
+import org.apache.nutch.plugin.TablePluggable;
+import org.apache.nutch.util.hbase.WebTableRow;
/** Extension point for DOM-based HTML parsers. Permits one to add additional
* metadata to HTML parses. All plugins found which implement this extension
* point are run sequentially on the parse.
*/
-public interface HtmlParseFilter extends Pluggable, Configurable {
+public interface HtmlParseFilter extends TablePluggable, Configurable {
/** The name of the extension point. */
final static String X_POINT_ID = HtmlParseFilter.class.getName();
/** Adds metadata or otherwise modifies a parse of HTML content, given
* the DOM tree of a page. */
- ParseResult filter(Content content, ParseResult parseResult, HTMLMetaTags metaTags, DocumentFragment doc);
+ Parse filter(String url, WebTableRow row, Parse parseResult,
+ HTMLMetaTags metaTags, DocumentFragment doc);
+
}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HtmlParseFilters.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HtmlParseFilters.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HtmlParseFilters.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/HtmlParseFilters.java Sun Aug 16 22:25:12 2009
@@ -17,11 +17,15 @@
package org.apache.nutch.parse;
+import java.util.Collection;
import java.util.HashMap;
+import java.util.HashSet;
-import org.apache.nutch.protocol.Content;
+import org.apache.nutch.parse.HTMLMetaTags;
import org.apache.nutch.plugin.*;
import org.apache.nutch.util.ObjectCache;
+import org.apache.nutch.util.hbase.HbaseColumn;
+import org.apache.nutch.util.hbase.WebTableRow;
import org.apache.hadoop.conf.Configuration;
import org.w3c.dom.DocumentFragment;
@@ -33,51 +37,59 @@
public HtmlParseFilters(Configuration conf) {
ObjectCache objectCache = ObjectCache.get(conf);
- this.htmlParseFilters = (HtmlParseFilter[]) objectCache.getObject(HtmlParseFilter.class.getName());
+ this.htmlParseFilters =
+ (HtmlParseFilter[]) objectCache.getObject(HtmlParseFilter.class.getName());
if (htmlParseFilters == null) {
HashMap<String, HtmlParseFilter> filters =
new HashMap<String, HtmlParseFilter>();
try {
- ExtensionPoint point = PluginRepository.get(conf).getExtensionPoint(HtmlParseFilter.X_POINT_ID);
+ ExtensionPoint point =
+ PluginRepository.get(conf).getExtensionPoint(HtmlParseFilter.X_POINT_ID);
if (point == null)
throw new RuntimeException(HtmlParseFilter.X_POINT_ID + " not found.");
Extension[] extensions = point.getExtensions();
for (int i = 0; i < extensions.length; i++) {
Extension extension = extensions[i];
- HtmlParseFilter parseFilter = (HtmlParseFilter) extension.getExtensionInstance();
+ HtmlParseFilter parseFilter =
+ (HtmlParseFilter) extension.getExtensionInstance();
if (!filters.containsKey(parseFilter.getClass().getName())) {
filters.put(parseFilter.getClass().getName(), parseFilter);
}
}
- HtmlParseFilter[] htmlParseFilters = filters.values().toArray(new HtmlParseFilter[filters.size()]);
+ HtmlParseFilter[] htmlParseFilters =
+ filters.values().toArray(new HtmlParseFilter[filters.size()]);
objectCache.setObject(HtmlParseFilter.class.getName(), htmlParseFilters);
} catch (PluginRuntimeException e) {
throw new RuntimeException(e);
}
- this.htmlParseFilters = (HtmlParseFilter[]) objectCache.getObject(HtmlParseFilter.class.getName());
+ this.htmlParseFilters =
+ (HtmlParseFilter[])objectCache.getObject(HtmlParseFilter.class.getName());
}
}
/** Run all defined filters. */
- public ParseResult filter(Content content, ParseResult parseResult, HTMLMetaTags metaTags, DocumentFragment doc) {
+ public Parse filter(String url, WebTableRow row, Parse parse,
+ HTMLMetaTags metaTags, DocumentFragment doc) {
// loop on each filter
- for (int i = 0 ; i < this.htmlParseFilters.length; i++) {
+ for (HtmlParseFilter htmlParseFilter : htmlParseFilters) {
// call filter interface
- parseResult =
- htmlParseFilters[i].filter(content, parseResult, metaTags, doc);
+ parse = htmlParseFilter.filter(url, row, parse, metaTags, doc);
// any failure on parse obj, return
- if (!parseResult.isSuccess()) {
- // TODO: What happens when parseResult.isEmpty() ?
- // Maybe clone parseResult and use parseResult as backup...
-
- // remove failed parse before return
- parseResult.filter();
- return parseResult;
+ if (!parse.getParseStatus().isSuccess()) {
+ return parse;
}
}
- return parseResult;
+ return parse;
+ }
+
+ public Collection<HbaseColumn> getColumns() {
+ Collection<HbaseColumn> columns = new HashSet<HbaseColumn>();
+ for (HtmlParseFilter htmlParseFilter : htmlParseFilters) {
+ columns.addAll(htmlParseFilter.getColumns());
+ }
+ return columns;
}
}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Outlink.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Outlink.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Outlink.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Outlink.java Sun Aug 16 22:25:12 2009
@@ -18,7 +18,6 @@
package org.apache.nutch.parse;
import java.io.*;
-import java.net.MalformedURLException;
import org.apache.hadoop.io.*;
@@ -30,7 +29,7 @@
public Outlink() {}
- public Outlink(String toUrl, String anchor) throws MalformedURLException {
+ public Outlink(String toUrl, String anchor) {
this.toUrl = toUrl;
if (anchor == null) anchor = "";
this.anchor = anchor;
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/OutlinkExtractor.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/OutlinkExtractor.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/OutlinkExtractor.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/OutlinkExtractor.java Sun Aug 16 22:25:12 2009
@@ -17,7 +17,6 @@
package org.apache.nutch.parse;
-import java.net.MalformedURLException;
import java.util.ArrayList;
import java.util.List;
@@ -110,11 +109,7 @@
}
result = matcher.getMatch();
url = result.group(0);
- try {
- outlinks.add(new Outlink(url, anchor));
- } catch (MalformedURLException mue) {
- LOG.warn("Invalid url: '" + url + "', skipping.");
- }
+ outlinks.add(new Outlink(url, anchor));
}
} catch (Exception ex) {
// if the matcher fails (perhaps a malformed URL) we just log it and move on
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Parse.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Parse.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Parse.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Parse.java Sun Aug 16 22:25:12 2009
@@ -1,34 +1,82 @@
-/**
- * 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.parse;
-/** The result of parsing a page's raw content.
- * @see Parser#getParse(Content)
- */
-public interface Parse {
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.nutch.parse.Outlink;
+import org.apache.nutch.parse.ParseStatus;
+
+public class Parse implements Writable {
+
+ private String text;
+ private String title;
+ private Outlink[] outlinks;
+ private ParseStatus parseStatus;
+
+ public Parse() {
+ }
- /** The textual content of the page. This is indexed, searched, and used when
- * generating snippets.*/
- String getText();
+ public Parse(String text, String title, Outlink[] outlinks,
+ ParseStatus parseStatus) {
+ this.text = text;
+ this.title = title;
+ this.outlinks = outlinks;
+ this.parseStatus = parseStatus;
+ }
+
+ public String getText() {
+ return text;
+ }
+
+ public String getTitle() {
+ return title;
+ }
- /** Other data extracted from the page. */
- ParseData getData();
+ public Outlink[] getOutlinks() {
+ return outlinks;
+ }
+
+ public ParseStatus getParseStatus() {
+ return parseStatus;
+ }
+
+ public void setText(String text) {
+ this.text = text;
+ }
+
+ public void setTitle(String title) {
+ this.title = title;
+ }
+
+ public void setOutlinks(Outlink[] outlinks) {
+ this.outlinks = outlinks;
+ }
+
+ public void setParseStatus(ParseStatus parseStatus) {
+ this.parseStatus = parseStatus;
+ }
- /** Indicates if the parse is coming from a url or a sub-url */
- boolean isCanonical();
+ public void readFields(DataInput in) throws IOException {
+ text = Text.readString(in);
+ title = Text.readString(in);
+ int numOutlinks = in.readInt();
+ outlinks = new Outlink[numOutlinks];
+ for (int i = 0; i < numOutlinks; i++) {
+ outlinks[i] = Outlink.read(in);
+ }
+ parseStatus = ParseStatus.read(in);
+ }
+
+ public void write(DataOutput out) throws IOException {
+ Text.writeString(out, text);
+ Text.writeString(out, title);
+ out.writeInt(outlinks.length);
+ for (Outlink outlink : outlinks) {
+ outlink.write(out);
+ }
+ parseStatus.write(out);
+ }
}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseData.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseData.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseData.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseData.java Sun Aug 16 22:25:12 2009
@@ -21,6 +21,7 @@
import java.util.*;
import org.apache.hadoop.io.*;
+import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
@@ -197,16 +198,20 @@
return buffer.toString();
}
- public static void main(String argv[]) throws Exception {
- String usage = "ParseData (-local | -dfs <namenode:port>) recno segment";
+ public static void main(String args[]) throws Exception {
+ String usage = "ParseData (-fs <local|namenode:port>) recno segment";
- if (argv.length < 3) {
+ if (args.length < 3) {
System.out.println("usage:" + usage);
return;
}
- Configuration conf = NutchConfiguration.create();
- FileSystem fs = FileSystem.parseArgs(argv, 0, conf);
+ GenericOptionsParser optParser =
+ new GenericOptionsParser(NutchConfiguration.create(), args);
+ String[] argv = optParser.getRemainingArgs();
+ Configuration conf = optParser.getConfiguration();
+
+ FileSystem fs = FileSystem.get(conf);
try {
int recno = Integer.parseInt(argv[0]);
String segment = argv[1];
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParsePluginList.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParsePluginList.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParsePluginList.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParsePluginList.java Sun Aug 16 22:25:12 2009
@@ -32,7 +32,7 @@
* @author mattmann
* @version 1.0
*/
-class ParsePluginList {
+public class ParsePluginList {
/* a map to link mimeType to an ordered list of parsing plugins */
private Map<String, List<String>> fMimeTypeToPluginMap = null;
@@ -49,7 +49,7 @@
aliases = new HashMap<String, String>();
}
- List<String> getPluginList(String mimeType) {
+ public List<String> getPluginList(String mimeType) {
return fMimeTypeToPluginMap.get(mimeType);
}
@@ -57,7 +57,7 @@
this.aliases = aliases;
}
- Map<String, String> getAliases() {
+ public Map<String, String> getAliases() {
return aliases;
}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParsePluginsReader.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParsePluginsReader.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParsePluginsReader.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParsePluginsReader.java Sun Aug 16 22:25:12 2009
@@ -50,7 +50,7 @@
* @author mattmann
* @version 1.0
*/
-class ParsePluginsReader {
+public class ParsePluginsReader {
/* our log stream */
public static final Log LOG = LogFactory.getLog(ParsePluginsReader.class);
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseStatus.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseStatus.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseStatus.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseStatus.java Sun Aug 16 22:25:12 2009
@@ -30,8 +30,6 @@
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.conf.Configuration;
-import org.apache.nutch.metadata.Metadata;
-
/**
* @author Andrzej Bialecki <ab@getopt.org>
@@ -201,12 +199,19 @@
return new EmptyParseImpl(this, conf);
}
+ /** A convenience method. Creates an empty ParseHbase instance,
+ * which returns this status.
+ */
+ public Parse getEmptyParseHbase(Configuration conf) {
+ return new Parse("", "", new Outlink[0], this);
+ }
+
/** A convenience method. Creates an empty ParseResult,
* which contains this status.
*/
- public ParseResult getEmptyParseResult(String url, Configuration conf) {
- return ParseResult.createParseResult(url, getEmptyParse(conf));
- }
+ //public ParseResult getEmptyParseResult(String url, Configuration conf) {
+ // return ParseResult.createParseResult(url, getEmptyParse(conf));
+ //}
public String toString() {
StringBuffer res = new StringBuffer();
@@ -267,25 +272,10 @@
return true;
}
- private static class EmptyParseImpl implements Parse {
-
- private ParseData data = null;
-
- public EmptyParseImpl(ParseStatus status, Configuration conf) {
- data = new ParseData(status, "", new Outlink[0],
- new Metadata(), new Metadata());
- }
-
- public ParseData getData() {
- return data;
- }
-
- public String getText() {
- return "";
- }
+ private static class EmptyParseImpl extends Parse {
- public boolean isCanonical() {
- return true;
+ public EmptyParseImpl(ParseStatus parseStatus, Configuration conf) {
+ super("", "", new Outlink[0], parseStatus);
}
}
}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseText.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseText.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseText.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseText.java Sun Aug 16 22:25:12 2009
@@ -19,6 +19,7 @@
import java.io.*;
import org.apache.hadoop.io.*;
+import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.conf.*;
import org.apache.nutch.util.NutchConfiguration;
@@ -79,16 +80,20 @@
return text;
}
- public static void main(String argv[]) throws Exception {
+ public static void main(String args[]) throws Exception {
String usage = "ParseText (-local | -dfs <namenode:port>) recno segment";
- if (argv.length < 3) {
+ if (args.length < 3) {
System.out.println("usage:" + usage);
return;
}
+
+ GenericOptionsParser optParser =
+ new GenericOptionsParser(NutchConfiguration.create(), args);
+ String[] argv = optParser.getRemainingArgs();
+ Configuration conf = optParser.getConfiguration();
- Configuration conf = NutchConfiguration.create();
- FileSystem fs = FileSystem.parseArgs(argv, 0, conf);
+ FileSystem fs = FileSystem.get(conf);
try {
int recno = Integer.parseInt(argv[0]);
String segment = argv[1];
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseUtil.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseUtil.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseUtil.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParseUtil.java Sun Aug 16 22:25:12 2009
@@ -21,12 +21,13 @@
import org.apache.commons.logging.LogFactory;
// Nutch Imports
-import org.apache.nutch.protocol.Content;
+import org.apache.nutch.parse.ParseException;
+import org.apache.nutch.parse.ParserNotFound;
+import org.apache.nutch.util.hbase.WebTableRow;
// Hadoop imports
import org.apache.hadoop.conf.Configuration;
-
/**
* A Utility class containing methods to simply perform parsing utilities such
* as iterating through a preferred list of {@link Parser}s to obtain
@@ -56,84 +57,37 @@
* returned. If the parse is unsuccessful, a message is logged to the
* <code>WARNING</code> level, and an empty parse is returned.
*
- * @param content The content to try and parse.
- * @return <key, {@link Parse}> pairs.
* @throws ParseException If no suitable parser is found to perform the parse.
*/
- public ParseResult parse(Content content) throws ParseException {
+ public Parse parse(String url, WebTableRow row) throws ParseException {
Parser[] parsers = null;
+ String contentType = row.getContentType();
+
try {
- parsers = this.parserFactory.getParsers(content.getContentType(),
- content.getUrl() != null ? content.getUrl():"");
+ parsers = this.parserFactory.getParsers(contentType, url);
} catch (ParserNotFound e) {
if (LOG.isWarnEnabled()) {
- LOG.warn("No suitable parser found when trying to parse content " + content.getUrl() +
- " of type " + content.getContentType());
+ LOG.warn("No suitable parser found when trying to parse content " + url +
+ " of type " + contentType);
}
throw new ParseException(e.getMessage());
}
- ParseResult parseResult = null;
+ Parse parse = null;
for (int i=0; i<parsers.length; i++) {
if (LOG.isDebugEnabled()) {
- LOG.debug("Parsing [" + content.getUrl() + "] with [" + parsers[i] + "]");
+ LOG.debug("Parsing [" + url + "] with [" + parsers[i] + "]");
}
- parseResult = parsers[i].getParse(content);
- if (parseResult != null && !parseResult.isEmpty())
- return parseResult;
+ parse = parsers[i].getParse(url, row);
+ if (parse != null)
+ return parse;
}
if (LOG.isWarnEnabled()) {
- LOG.warn("Unable to successfully parse content " + content.getUrl() +
- " of type " + content.getContentType());
+ LOG.warn("Unable to successfully parse content " + url +
+ " of type " + contentType);
}
return null;
}
-
- /**
- * Method parses a {@link Content} object using the {@link Parser} specified
- * by the parameter <code>extId</code>, i.e., the Parser's extension ID.
- * If a suitable {@link Parser} is not found, then a <code>WARNING</code>
- * level message is logged, and a ParseException is thrown. If the parse is
- * uncessful for any other reason, then a <code>WARNING</code> level
- * message is logged, and a <code>ParseStatus.getEmptyParse()</code> is
- * returned.
- *
- * @param extId The extension implementation ID of the {@link Parser} to use
- * to parse the specified content.
- * @param content The content to parse.
- *
- * @return <key, {@link Parse}> pairs if the parse is successful, otherwise,
- * a single <key, <code>ParseStatus.getEmptyParse()</code>> pair.
- *
- * @throws ParseException If there is no suitable {@link Parser} found
- * to perform the parse.
- */
- public ParseResult parseByExtensionId(String extId, Content content)
- throws ParseException {
- Parser p = null;
-
- try {
- p = this.parserFactory.getParserById(extId);
- } catch (ParserNotFound e) {
- if (LOG.isWarnEnabled()) {
- LOG.warn("No suitable parser found when trying to parse content " + content.getUrl() +
- " of type " + content.getContentType());
- }
- throw new ParseException(e.getMessage());
- }
-
- ParseResult parseResult = p.getParse(content);
- if (parseResult != null && !parseResult.isEmpty()) {
- return parseResult;
- } else {
- if (LOG.isWarnEnabled()) {
- LOG.warn("Unable to successfully parse content " + content.getUrl() +
- " of type " + content.getContentType());
- }
- return null;
- }
- }
-
}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Parser.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Parser.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Parser.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/Parser.java Sun Aug 16 22:25:12 2009
@@ -21,36 +21,27 @@
import org.apache.hadoop.conf.Configurable;
// Nutch imports
-import org.apache.nutch.plugin.Pluggable;
-import org.apache.nutch.protocol.Content;
+import org.apache.nutch.plugin.TablePluggable;
+import org.apache.nutch.util.hbase.WebTableRow;
/** A parser for content generated by a {@link org.apache.nutch.protocol.Protocol}
* implementation. This interface is implemented by extensions. Nutch's core
* contains no page parsing code.
*/
-public interface Parser extends Pluggable, Configurable {
+public interface Parser extends TablePluggable, Configurable {
/** The name of the extension point. */
public final static String X_POINT_ID = Parser.class.getName();
/**
* <p>
- * This method parses the given content and returns a map of
- * <key, parse> pairs. {@link Parse} instances will be persisted
- * under the given key.
- * </p>
- * <p>
- * Note: Meta-redirects should be followed only when they are coming from
- * the original URL. That is: <br>
- * Assume fetcher is in parsing mode and is currently processing
- * foo.bar.com/redirect.html. If this url contains a meta redirect
- * to another url, fetcher should only follow the redirect if the map
- * contains an entry of the form <"foo.bar.com/redirect.html",
- * {@link Parse} with a {@link ParseStatus} indicating the redirect>.
+ * This method takes a Hbase table row and returns a {@link Parse} instance
* </p>
*
- * @param c Content to be parsed
- * @return a map containing <key, parse> pairs
- * @since NUTCH-443
+ * @param url Page's URL
+ * @param row url's row. Parse plugins can add their updates here. They
+ * will be reflected in hbase. Plugins must not update for items already
+ * in ParseHbase
+ * @return Parse
*/
- ParseResult getParse(Content c);
+ Parse getParse(String url, WebTableRow row);
}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParserFactory.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParserFactory.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParserFactory.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/ParserFactory.java Sun Aug 16 22:25:12 2009
@@ -18,10 +18,10 @@
// JDK imports
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Iterator;
+import java.util.Collection;
+import java.util.HashSet;
import java.util.List;
-import java.util.Vector;
+import java.util.Set;
// Commons Logging imports
import org.apache.commons.logging.Log;
@@ -31,6 +31,9 @@
import org.apache.hadoop.conf.Configuration;
// Nutch imports
+import org.apache.nutch.parse.ParsePluginList;
+import org.apache.nutch.parse.ParsePluginsReader;
+import org.apache.nutch.parse.ParserNotFound;
import org.apache.nutch.plugin.Extension;
import org.apache.nutch.plugin.ExtensionPoint;
import org.apache.nutch.plugin.PluginRuntimeException;
@@ -38,6 +41,7 @@
import org.apache.nutch.util.LogUtil;
import org.apache.nutch.util.MimeUtil;
import org.apache.nutch.util.ObjectCache;
+import org.apache.nutch.util.hbase.HbaseColumn;
/** Creates and caches {@link Parser} plugins.*/
@@ -49,7 +53,8 @@
public static final String DEFAULT_PLUGIN = "*";
/** Empty extension list for caching purposes. */
- private final List EMPTY_EXTENSION_LIST = Collections.EMPTY_LIST;
+ private final List<Extension> EMPTY_EXTENSION_LIST =
+ new ArrayList<Extension>();
private Configuration conf;
private ExtensionPoint extensionPoint;
@@ -121,9 +126,8 @@
throw new ParserNotFound(url, contentType);
}
- parsers = new Vector<Parser>(parserExts.size());
- for (Iterator i=parserExts.iterator(); i.hasNext(); ){
- Extension ext = (Extension) i.next();
+ parsers = new ArrayList<Parser>(parserExts.size());
+ for (Extension ext : parserExts){
Parser p = null;
try {
//check to see if we've cached this parser instance yet
@@ -207,6 +211,20 @@
}
}
+ public Collection<HbaseColumn> getColumns() {
+ Set<HbaseColumn> columns = new HashSet<HbaseColumn>();
+ Extension[] extensions = this.extensionPoint.getExtensions();
+ for (Extension ext : extensions) {
+ try {
+ Parser parser = (Parser) ext.getExtensionInstance();
+ columns.addAll(parser.getColumns());
+ } catch (PluginRuntimeException e) {
+ // ignore
+ }
+ }
+ return columns;
+ }
+
/**
* Finds the best-suited parse plugin for a given contentType.
*
@@ -214,6 +232,7 @@
* @return a list of extensions to be used for this contentType.
* If none, returns <code>null</code>.
*/
+ @SuppressWarnings("unchecked")
protected List<Extension> getExtensions(String contentType) {
ObjectCache objectCache = ObjectCache.get(conf);
@@ -221,7 +240,6 @@
String type = null;
type = MimeUtil.cleanMimeType(contentType);
-
List<Extension> extensions = (List<Extension>) objectCache.getObject(type);
// Just compare the reference:
Added: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/TableParser.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/TableParser.java?rev=804789&view=auto
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/TableParser.java (added)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/parse/TableParser.java Sun Aug 16 22:25:12 2009
@@ -0,0 +1,276 @@
+package org.apache.nutch.parse;
+
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.filter.FilterList;
+import org.apache.hadoop.hbase.filter.ValueFilter;
+import org.apache.hadoop.hbase.filter.FilterList.Operator;
+import org.apache.hadoop.hbase.filter.ValueFilter.CompareOp;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.mapreduce.IdentityTableReducer;
+import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
+import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
+import org.apache.hadoop.hbase.mapreduce.TableMapper;
+import org.apache.hadoop.hbase.util.Bytes;
+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.CrawlDatumHbase;
+import org.apache.nutch.crawl.Signature;
+import org.apache.nutch.crawl.SignatureFactory;
+import org.apache.nutch.fetcher.Fetcher;
+import org.apache.nutch.net.URLFilterException;
+import org.apache.nutch.net.URLFilters;
+import org.apache.nutch.net.URLNormalizers;
+import org.apache.nutch.parse.Outlink;
+import org.apache.nutch.parse.ParseStatus;
+import org.apache.nutch.util.NutchConfiguration;
+import org.apache.nutch.util.NutchJob;
+import org.apache.nutch.util.URLUtil;
+import org.apache.nutch.util.hbase.HbaseColumn;
+import org.apache.nutch.util.hbase.WebTableRow;
+import org.apache.nutch.util.hbase.WebTableColumns;
+import org.apache.nutch.util.hbase.TableUtil;
+
+public class TableParser
+extends TableMapper<ImmutableBytesWritable, ImmutableBytesWritable>
+implements Tool {
+
+ public static final Log LOG = LogFactory.getLog(TableParser.class);
+
+ public static final String PARSE_MARK_STR = "__prsmrk__";
+
+ public static final byte[] PARSE_MARK = Bytes.toBytes("__prsmrk__");
+
+ private static final Collection<HbaseColumn> COLUMNS = new HashSet<HbaseColumn>();
+
+ private Configuration conf;
+
+ static {
+ COLUMNS.add(new HbaseColumn(WebTableColumns.STATUS));
+ COLUMNS.add(new HbaseColumn(WebTableColumns.CONTENT));
+ COLUMNS.add(new HbaseColumn(WebTableColumns.CONTENT_TYPE));
+ COLUMNS.add(new HbaseColumn(WebTableColumns.SIGNATURE));
+ COLUMNS.add(new HbaseColumn(WebTableColumns.METADATA, Fetcher.FETCH_MARK));
+ }
+
+ private ParseUtil parseUtil;
+ private Signature sig;
+ private URLFilters filters;
+ private URLNormalizers normalizers;
+ private int maxOutlinks;
+ private boolean ignoreExternalLinks;
+ private HTable table;
+
+ @Override
+ public void setup(Context context) throws IOException {
+ Configuration conf = context.getConfiguration();
+ parseUtil = new ParseUtil(conf);
+ sig = SignatureFactory.getSignature(conf);
+ filters = new URLFilters(conf);
+ normalizers = new URLNormalizers(conf, URLNormalizers.SCOPE_OUTLINK);
+ final int maxOutlinksPerPage = conf.getInt("db.max.outlinks.per.page", 100);
+ maxOutlinks = (maxOutlinksPerPage < 0) ? Integer.MAX_VALUE
+ : maxOutlinksPerPage;
+ ignoreExternalLinks = conf.getBoolean("db.ignore.external.links", false);
+ table = new HTable(conf.get(TableInputFormat.INPUT_TABLE));
+ }
+
+ @Override
+ public void map(ImmutableBytesWritable key, Result result, Context context)
+ throws IOException {
+ final WebTableRow row = new WebTableRow(result);
+ final String url = TableUtil.unreverseUrl(Bytes.toString(key.get()));
+ final byte status = row.getStatus();
+ if (status != CrawlDatumHbase.STATUS_FETCHED) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Skipping " + url + " as status is: " +
+ CrawlDatumHbase.getName(status));
+ }
+ return;
+ }
+
+ Parse parse;
+ try {
+ parse = parseUtil.parse(url, row);
+ } catch (final Exception e) {
+ LOG.warn("Error parsing: " + url + ": " + StringUtils.stringifyException(e));
+ return;
+ }
+
+ final byte[] signature = sig.calculate(row, parse);
+
+ final ParseStatus pstatus = parse.getParseStatus();
+ row.setParseStatus(pstatus);
+ if (pstatus.isSuccess()) {
+ if (pstatus.getMinorCode() == ParseStatus.SUCCESS_REDIRECT) {
+ String newUrl = pstatus.getMessage();
+ final int refreshTime = Integer.parseInt(pstatus.getArgs()[1]);
+ newUrl = normalizers.normalize(newUrl,
+ URLNormalizers.SCOPE_FETCHER);
+ try {
+ newUrl = filters.filter(newUrl);
+ } catch (URLFilterException e) {
+ return; // TODO: is this correct
+ }
+ if (newUrl == null || newUrl.equals(url)) {
+ final String reprUrl = URLUtil.chooseRepr(url, newUrl,
+ refreshTime < Fetcher.PERM_REFRESH_TIME);
+ try {
+ final String reversedUrl = TableUtil.reverseUrl(reprUrl);
+ final ImmutableBytesWritable newKey =
+ new ImmutableBytesWritable(reversedUrl.getBytes());
+ final Put put = new Put(newKey.get());
+ if (!reprUrl.equals(url)) {
+ put.add(WebTableColumns.REPR_URL, null, Bytes.toBytes(reprUrl));
+ }
+ put.add(WebTableColumns.METADATA,
+ Fetcher.REDIRECT_DISCOVERED, TableUtil.YES_VAL);
+ table.put(put);
+ } catch (MalformedURLException e) {
+ // ignore
+ }
+ }
+ } else {
+ row.setText(parse.getText());
+ row.setTitle(parse.getTitle());
+ final byte[] prevSig = row.getSignature();
+ if (prevSig != null) {
+ row.setPrevSignature(prevSig);
+ }
+ row.setSignature(signature);
+ row.deleteAllOutlinks();
+ final Outlink[] outlinks = parse.getOutlinks();
+ final int count = 0;
+ String fromHost;
+ if (ignoreExternalLinks) {
+ try {
+ fromHost = new URL(url).getHost().toLowerCase();
+ } catch (final MalformedURLException e) {
+ fromHost = null;
+ }
+ } else {
+ fromHost = null;
+ }
+ for (int i = 0; count < maxOutlinks && i < outlinks.length; i++) {
+ String toUrl = outlinks[i].getToUrl();
+ toUrl = normalizers.normalize(toUrl, URLNormalizers.SCOPE_OUTLINK);
+ try {
+ toUrl = filters.filter(toUrl);
+ } catch (final URLFilterException e) {
+ continue;
+ }
+ if (toUrl == null) {
+ continue;
+ }
+ String toHost;
+ if (ignoreExternalLinks) {
+ try {
+ toHost = new URL(toUrl).getHost().toLowerCase();
+ } catch (final MalformedURLException e) {
+ toHost = null;
+ }
+ if (toHost == null || !toHost.equals(fromHost)) { // external links
+ continue; // skip it
+ }
+ }
+
+ row.addOutlink(new Outlink(toUrl, outlinks[i].getAnchor()));
+ }
+ row.putMeta(PARSE_MARK, TableUtil.YES_VAL);
+ }
+ }
+
+ row.makeRowMutation().commit(table);
+ }
+
+ @Override
+ public void cleanup(Context context) throws IOException {
+ table.close();
+ }
+
+ private Scan makeScan(Job job, boolean restart) {
+ Configuration conf = job.getConfiguration();
+ final Collection<HbaseColumn> columns = new HashSet<HbaseColumn>(COLUMNS);
+ final ParserFactory parserFactory = new ParserFactory(conf);
+ columns.addAll(parserFactory.getColumns());
+ columns.addAll(SignatureFactory.getColumns(conf));
+ final HtmlParseFilters parseFilters = new HtmlParseFilters(conf);
+ columns.addAll(parseFilters.getColumns());
+ Scan scan = TableUtil.createScanFromColumns(columns);
+ List<Filter> filters = new ArrayList<Filter>();
+ if (!restart) {
+ filters.add(new ValueFilter(WebTableColumns.METADATA, PARSE_MARK,
+ CompareOp.NOT_EQUAL, TableUtil.YES_VAL, false));
+ }
+ filters.add(new ValueFilter(WebTableColumns.METADATA, Fetcher.FETCH_MARK,
+ CompareOp.EQUAL, TableUtil.YES_VAL, true));
+ scan.setFilter(new FilterList(Operator.MUST_PASS_ALL, filters));
+ return scan;
+ }
+
+
+ @Override
+ public Configuration getConf() {
+ return conf;
+ }
+
+ @Override
+ public void setConf(Configuration conf) {
+ this.conf = conf;
+ }
+
+ public void parse(String table, boolean restart) throws Exception {
+
+ LOG.info("TableParser: starting");
+ LOG.info("TableParser: segment: " + table);
+
+ final Job job = new NutchJob(getConf(), "parse " + table);
+
+ Scan scan = makeScan(job, restart);
+ TableMapReduceUtil.initTableMapperJob(table, scan,
+ TableParser.class, ImmutableBytesWritable.class,
+ ImmutableBytesWritable.class, job);
+ TableMapReduceUtil.initTableReducerJob(table, IdentityTableReducer.class, job);
+ job.waitForCompletion(true);
+ LOG.info("TableParser: done");
+ }
+
+ public int run(String[] args) throws Exception {
+ final String usage = "Usage: TableParser <webtable> [-restart]";
+
+ if (args.length == 0) {
+ System.err.println(usage);
+ System.exit(-1);
+ }
+
+ boolean restart = false;
+ if (args.length >= 2 && "-restart".equals(args[1])) {
+ restart = true;
+ }
+
+ parse(args[0], restart);
+ return 0;
+ }
+
+ public static void main(String[] args) throws Exception {
+ final int res = ToolRunner.run(NutchConfiguration.create(), new TableParser(), args);
+ System.exit(res);
+ }
+
+}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/plugin/Pluggable.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/plugin/Pluggable.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/plugin/Pluggable.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/plugin/Pluggable.java Sun Aug 16 22:25:12 2009
@@ -1,21 +1,6 @@
-/*
- * 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.plugin;
+
/**
* Defines the capability of a class to be plugged into Nutch.
* This is a common interface that must be implemented by all
@@ -28,5 +13,4 @@
* plugin package description</a>
*/
public interface Pluggable {
-
}
Added: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/plugin/TablePluggable.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/plugin/TablePluggable.java?rev=804789&view=auto
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/plugin/TablePluggable.java (added)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/plugin/TablePluggable.java Sun Aug 16 22:25:12 2009
@@ -0,0 +1,15 @@
+package org.apache.nutch.plugin;
+
+import java.util.Collection;
+import java.util.HashSet;
+
+import org.apache.nutch.util.hbase.HbaseColumn;
+
+public interface TablePluggable extends Pluggable {
+
+ public static final Collection<HbaseColumn> EMPTY_COLUMNS =
+ new HashSet<HbaseColumn>();
+
+ public Collection<HbaseColumn> getColumns();
+
+}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/Content.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/Content.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/Content.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/Content.java Sun Aug 16 22:25:12 2009
@@ -35,6 +35,7 @@
import org.apache.hadoop.io.UTF8;
import org.apache.hadoop.io.VersionMismatchException;
import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.util.GenericOptionsParser;
//Nutch imports
import org.apache.nutch.metadata.Metadata;
@@ -85,6 +86,27 @@
this.mimeTypes = new MimeUtil(conf);
this.contentType = getContentType(contentType, url, content);
}
+
+ public Content(String url, String base, byte[] content, String contentType,
+ Metadata metadata, MimeUtil mimeTypes) {
+
+ if (url == null)
+ throw new IllegalArgumentException("null url");
+ if (base == null)
+ throw new IllegalArgumentException("null base");
+ if (content == null)
+ throw new IllegalArgumentException("null content");
+ if (metadata == null)
+ throw new IllegalArgumentException("null metadata");
+
+ this.url = url;
+ this.base = base;
+ this.content = content;
+ this.metadata = metadata;
+
+ this.mimeTypes = mimeTypes;
+ this.contentType = getContentType(contentType, url, content);
+ }
private final void readFieldsCompressed(DataInput in) throws IOException {
byte oldVersion = in.readByte();
@@ -247,16 +269,21 @@
}
- public static void main(String argv[]) throws Exception {
+ public static void main(String args[]) throws Exception {
String usage = "Content (-local | -dfs <namenode:port>) recno segment";
- if (argv.length < 3) {
+ if (args.length < 3) {
System.out.println("usage:" + usage);
return;
}
- Configuration conf = NutchConfiguration.create();
- FileSystem fs = FileSystem.parseArgs(argv, 0, conf);
+
+ GenericOptionsParser optParser =
+ new GenericOptionsParser(NutchConfiguration.create(), args);
+ String[] argv = optParser.getRemainingArgs();
+ Configuration conf = optParser.getConfiguration();
+
+ FileSystem fs = FileSystem.get(conf);
try {
int recno = Integer.parseInt(argv[0]);
String segment = argv[1];
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/Protocol.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/Protocol.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/Protocol.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/Protocol.java Sun Aug 16 22:25:12 2009
@@ -18,16 +18,18 @@
package org.apache.nutch.protocol;
// Hadoop imports
+
import org.apache.hadoop.conf.Configurable;
-import org.apache.hadoop.io.Text;
// Nutch imports
-import org.apache.nutch.crawl.CrawlDatum;
-import org.apache.nutch.plugin.Pluggable;
-
+import org.apache.nutch.plugin.TablePluggable;
+import org.apache.nutch.protocol.ProtocolOutput;
+import org.apache.nutch.protocol.RobotRules;
+import org.apache.nutch.util.hbase.WebTableRow;
+import org.apache.nutch.util.hbase.WebTableRow;
/** A retriever of url content. Implemented by protocol extensions. */
-public interface Protocol extends Pluggable, Configurable {
+public interface Protocol extends TablePluggable, Configurable {
/** The name of the extension point. */
public final static String X_POINT_ID = Protocol.class.getName();
@@ -51,13 +53,13 @@
/** Returns the {@link Content} for a fetchlist entry.
*/
- ProtocolOutput getProtocolOutput(Text url, CrawlDatum datum);
+ ProtocolOutput getProtocolOutput(String url, WebTableRow row);
/**
* Retrieve robot rules applicable for this url.
* @param url url to check
- * @param datum page datum
+ * @param row Row
* @return robot rules (specific for this url or default), never null
*/
- RobotRules getRobotRules(Text url, CrawlDatum datum);
+ RobotRules getRobotRules(String url, WebTableRow row);
}
Modified: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/ProtocolFactory.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/ProtocolFactory.java?rev=804789&r1=804788&r2=804789&view=diff
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/ProtocolFactory.java (original)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/protocol/ProtocolFactory.java Sun Aug 16 22:25:12 2009
@@ -19,13 +19,17 @@
import java.net.URL;
import java.net.MalformedURLException;
+import java.util.Collection;
+import java.util.HashSet;
// Commons Logging imports
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.nutch.plugin.*;
+import org.apache.nutch.protocol.ProtocolNotFound;
import org.apache.nutch.util.ObjectCache;
+import org.apache.nutch.util.hbase.HbaseColumn;
import org.apache.hadoop.conf.Configuration;
@@ -100,7 +104,6 @@
for (int i = 0; i < extensions.length; i++) {
Extension extension = extensions[i];
-
if (contains(name, extension.getAttribute("protocolName")))
return extension;
}
@@ -115,4 +118,18 @@
return false;
}
+ public Collection<HbaseColumn> getColumnSet() {
+ Collection<HbaseColumn> columns = new HashSet<HbaseColumn>();
+ for (Extension extension : this.extensionPoint.getExtensions()) {
+ Protocol protocol;
+ try {
+ protocol = (Protocol) extension.getExtensionInstance();
+ columns.addAll(protocol.getColumns());
+ } catch (PluginRuntimeException e) {
+ // ignore
+ }
+ }
+ return columns;
+ }
+
}
Added: lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/scoring/ScoreDatum.java
URL: http://svn.apache.org/viewvc/lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/scoring/ScoreDatum.java?rev=804789&view=auto
==============================================================================
--- lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/scoring/ScoreDatum.java (added)
+++ lucene/nutch/branches/nutchbase/src/java/org/apache/nutch/scoring/ScoreDatum.java Sun Aug 16 22:25:12 2009
@@ -0,0 +1,89 @@
+package org.apache.nutch.scoring;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableUtils;
+
+public class ScoreDatum implements Writable {
+
+ private float score;
+ private String url;
+ private String anchor;
+ private Map<String, byte[]> metaData = new HashMap<String, byte[]>();
+
+ public ScoreDatum() { }
+
+ public ScoreDatum(float score, String url, String anchor) {
+ this.score = score;
+ this.url = url;
+ this.anchor = anchor;
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ score = in.readFloat();
+ url = Text.readString(in);
+ anchor = Text.readString(in);
+ metaData.clear();
+
+ int size = WritableUtils.readVInt(in);
+ for (int i = 0; i < size; i++) {
+ String key = Text.readString(in);
+ byte[] value = Bytes.readByteArray(in);
+ metaData.put(key, value);
+ }
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeFloat(score);
+ Text.writeString(out, url);
+ Text.writeString(out, anchor);
+
+ WritableUtils.writeVInt(out, metaData.size());
+ for (Entry<String, byte[]> e : metaData.entrySet()) {
+ Text.writeString(out, e.getKey());
+ Bytes.writeByteArray(out, e.getValue());
+ }
+ }
+
+ public byte[] getMeta(String key) {
+ return metaData.get(key);
+ }
+
+ public void setMeta(String key, byte[] value) {
+ metaData.put(key, value);
+ }
+
+ public byte[] deleteMeta(String key) {
+ return metaData.remove(key);
+ }
+
+ public float getScore() {
+ return score;
+ }
+
+ public void setScore(float score) {
+ this.score = score;
+ }
+
+ public String getUrl() {
+ return url;
+ }
+
+ public void setUrl(String url) {
+ this.url = url;
+ }
+
+ public String getAnchor() {
+ return anchor;
+ }
+}