NUTCH-2481 HostDatum deltas(previous step statistics) and Metadata expressions

conf/nutch-default.xml

@@ -2596,6 +2596,15 @@ visit https://wiki.apache.org/nutch/SimilarityScoringFilter-->
   </description>
 </property>
 
+<property>
+  <name>hostdb.deltaExpression</name>
+  <value></value>
+  <description>
+    The expression for calculation of the delta statistics, the differences between of value of hostdb after update(currentHostDatum) and the value before(previousHostDatum). The return value in the KeyValuePair(String,Number) or KeyValuePair(String, String) format is written to the metadata of the hostdb.
+For example, {return new ("javafx.util.Pair","FetchedDelta", currentHostDatum.fetched - previousHostDatum.fetched);}
+  </description>
+</property>
+
 <!-- publisher properties 
       Do not forget to add the name of your publisher implementation 
       in plugin.includes ex- publish-rabbitmq -->

src/java/org/apache/nutch/hostdb/ReadHostDb.java

@@ -21,46 +21,41 @@
 import java.text.SimpleDateFormat;
 import java.util.Map;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.commons.jexl2.Expression;
+import org.apache.commons.jexl2.JexlContext;
+import org.apache.commons.jexl2.MapContext;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.io.FloatWritable;
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.SequenceFile;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.mapred.SequenceFileOutputFormat;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.Reducer;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
 import org.apache.nutch.util.NutchConfiguration;
-import org.apache.nutch.util.StringUtil;
 import org.apache.nutch.util.TimingUtil;
-import org.apache.nutch.util.URLUtil;
-
-import org.apache.commons.jexl2.JexlContext;
-import org.apache.commons.jexl2.Expression;
-import org.apache.commons.jexl2.JexlEngine;
-import org.apache.commons.jexl2.MapContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
- * @see <a href='http://commons.apache.org/proper/commons-jexl/reference/syntax.html'>Commons</a>
+ * @see <a href=
+ *      'http://commons.apache.org/proper/commons-jexl/reference/syntax.html'>Commons</a
+ *      >
  */
 public class ReadHostDb extends Configured implements Tool {
 
-  private static final Logger LOG = LoggerFactory
-      .getLogger(MethodHandles.lookup().lookupClass());
+  private static final Logger LOG = LoggerFactory.getLogger(MethodHandles
+      .lookup().lookupClass());
 
   public static final String HOSTDB_DUMP_HEADER = "hostdb.dump.field.header";
   public static final String HOSTDB_DUMP_HOSTNAMES = "hostdb.dump.hostnames";
@@ -75,33 +70,33 @@ static class ReadHostDbMapper extends Mapper<Text, HostDatum, Text, Text> {
     protected Expression expr = null;
 
     public void setup(Context context) {
-      dumpHomepages = context.getConfiguration().getBoolean(HOSTDB_DUMP_HOMEPAGES, false);
-      dumpHostnames = context.getConfiguration().getBoolean(HOSTDB_DUMP_HOSTNAMES, false);
-      fieldHeader = context.getConfiguration().getBoolean(HOSTDB_DUMP_HEADER, true);
+      dumpHomepages = context.getConfiguration().getBoolean(
+          HOSTDB_DUMP_HOMEPAGES, false);
+      dumpHostnames = context.getConfiguration().getBoolean(
+          HOSTDB_DUMP_HOSTNAMES, false);
+      fieldHeader = context.getConfiguration().getBoolean(HOSTDB_DUMP_HEADER,
+          true);
       String expr = context.getConfiguration().get(HOSTDB_FILTER_EXPRESSION);
       if (expr != null) {
-        // Create or retrieve a JexlEngine
-        JexlEngine jexl = new JexlEngine();
-
-        // Dont't be silent and be strict
-        jexl.setSilent(true);
-        jexl.setStrict(true);
-
-        // Create an expression object
-        this.expr = jexl.createExpression(expr);
+        this.expr = org.apache.nutch.util.JexlUtil.parseExpression(expr);
       }
     }
 
-    public void map(Text key, HostDatum datum, Context context) throws IOException, InterruptedException {
+    public void map(Text key, HostDatum datum, Context context)
+        throws IOException, InterruptedException {
       if (fieldHeader && !dumpHomepages && !dumpHostnames) {
-        context.write(new Text("hostname"), new Text("unfetched\tfetched\tgone\tredirTemp\tredirPerm\tredirSum\tok\tnumRecords\tdnsFail\tcnxFail\tsumFail\tscore\tlastCheck\thomepage\tmetadata"));
+        context
+            .write(
+                new Text("hostname"),
+                new Text(
+                    "unfetched\tfetched\tgone\tredirTemp\tredirPerm\tredirSum\tok\tnumRecords\tdnsFail\tcnxFail\tsumFail\tscore\tlastCheck\thomepage\tmetadata"));
         fieldHeader = false;
       }
-      
+
       if (expr != null) {
         // Create a context and add data
         JexlContext jcontext = new MapContext();
-        
+
         // Set some fixed variables
         jcontext.set("unfetched", datum.getUnfetched());
         jcontext.set("fetched", datum.getFetched());
@@ -114,24 +109,25 @@ public void map(Text key, HostDatum datum, Context context) throws IOException,
         jcontext.set("numRecords", datum.numRecords());
         jcontext.set("dnsFailures", datum.getDnsFailures());
         jcontext.set("connectionFailures", datum.getConnectionFailures());
-        
+
         // Set metadata variables
-        for (Map.Entry<Writable, Writable> entry : datum.getMetaData().entrySet()) {
+        for (Map.Entry<Writable, Writable> entry : datum.getMetaData()
+            .entrySet()) {
           Object value = entry.getValue();
-          
+
           if (value instanceof FloatWritable) {
-            FloatWritable fvalue = (FloatWritable)value;
-            Text tkey = (Text)entry.getKey();
+            FloatWritable fvalue = (FloatWritable) value;
+            Text tkey = (Text) entry.getKey();
             jcontext.set(tkey.toString(), fvalue.get());
           }
-          
+
           if (value instanceof IntWritable) {
-            IntWritable ivalue = (IntWritable)value;
-            Text tkey = (Text)entry.getKey();
+            IntWritable ivalue = (IntWritable) value;
+            Text tkey = (Text) entry.getKey();
             jcontext.set(tkey.toString(), ivalue.get());
           }
         }
-        
+
         // Filter this record if evaluation did not pass
         try {
           if (!Boolean.TRUE.equals(expr.evaluate(jcontext))) {
@@ -141,35 +137,38 @@ public void map(Text key, HostDatum datum, Context context) throws IOException,
           LOG.info(e.toString() + " for " + key.toString());
         }
       }
-      
+
       if (dumpHomepages) {
         if (datum.hasHomepageUrl()) {
           context.write(new Text(datum.getHomepageUrl()), emptyText);
         }
         return;
       }
-      
+
       if (dumpHostnames) {
         context.write(key, emptyText);
         return;
       }
-      
+
       // Write anyway
       context.write(key, new Text(datum.toString()));
     }
   }
 
-  // Todo, reduce unknown hosts to single unknown domain if possible. Enable via configuration
+  // Todo, reduce unknown hosts to single unknown domain if possible. Enable via
+  // configuration
   // host_a.example.org,host_a.example.org ==> example.org
-//   static class ReadHostDbReducer extends Reduce<Text, Text, Text, Text> {
-//     public void setup(Context context) { }
-//
-//     public void reduce(Text domain, Iterable<Text> hosts, Context context) throws IOException, InterruptedException {
-//
-//     }
-//   }
-
-  private void readHostDb(Path hostDb, Path output, boolean dumpHomepages, boolean dumpHostnames, String expr) throws Exception {
+  // static class ReadHostDbReducer extends Reduce<Text, Text, Text, Text> {
+  // public void setup(Context context) { }
+  //
+  // public void reduce(Text domain, Iterable<Text> hosts, Context context)
+  // throws IOException, InterruptedException {
+  //
+  // }
+  // }
+
+  private void readHostDb(Path hostDb, Path output, boolean dumpHomepages,
+      boolean dumpHostnames, String expr) throws Exception {
     SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
     long start = System.currentTimeMillis();
     LOG.info("ReadHostDb: starting at " + sdf.format(start));
@@ -182,7 +181,7 @@ private void readHostDb(Path hostDb, Path output, boolean dumpHomepages, boolean
     }
     conf.setBoolean("mapreduce.fileoutputcommitter.marksuccessfuljobs", false);
     conf.set("mapred.textoutputformat.separator", "\t");
-    
+
     Job job = Job.getInstance(conf);
     job.setJobName("ReadHostDb");
     job.setJarByClass(ReadHostDb.class);
@@ -217,40 +216,44 @@ private void readHostDb(Path hostDb, Path output, boolean dumpHomepages, boolean
     }
 
     long end = System.currentTimeMillis();
-    LOG.info("ReadHostDb: finished at " + sdf.format(end) + ", elapsed: " + TimingUtil.elapsedTime(start, end));
+    LOG.info("ReadHostDb: finished at " + sdf.format(end) + ", elapsed: "
+        + TimingUtil.elapsedTime(start, end));
   }
-  
+
   private void getHostDbRecord(Path hostDb, String host) throws Exception {
     Configuration conf = getConf();
-    SequenceFile.Reader[] readers = SequenceFileOutputFormat.getReaders(conf, hostDb);
+    SequenceFile.Reader[] readers = SequenceFileOutputFormat.getReaders(conf,
+        hostDb);
 
     Class<?> keyClass = readers[0].getKeyClass();
     Class<?> valueClass = readers[0].getValueClass();
-    
+
     if (!keyClass.getName().equals("org.apache.hadoop.io.Text"))
       throw new IOException("Incompatible key (" + keyClass.getName() + ")");
-      
+
     Text key = (Text) keyClass.newInstance();
     HostDatum value = (HostDatum) valueClass.newInstance();
-    
+
     for (int i = 0; i < readers.length; i++) {
       while (readers[i].next(key, value)) {
         if (host.equals(key.toString())) {
           System.out.println(value.toString());
         }
       }
       readers[i].close();
-    }    
+    }
   }
 
   public static void main(String args[]) throws Exception {
-    int res = ToolRunner.run(NutchConfiguration.create(), new ReadHostDb(), args);
+    int res = ToolRunner.run(NutchConfiguration.create(), new ReadHostDb(),
+        args);
     System.exit(res);
   }
 
   public int run(String[] args) throws Exception {
     if (args.length < 2) {
-      System.err.println("Usage: ReadHostDb <hostdb> [-get <url>] [<output> [-dumpHomepages | -dumpHostnames | -expr <expr.>]]");
+      System.err
+          .println("Usage: ReadHostDb <hostdb> [-get <url>] [<output> [-dumpHomepages | -dumpHostnames | -expr <expr.>]]");
       return -1;
     }
 
@@ -270,7 +273,7 @@ public int run(String[] args) throws Exception {
       }
       if (args[i].equals("-get")) {
         get = args[i + 1];
-        LOG.info("ReadHostDb: get: "+ get);
+        LOG.info("ReadHostDb: get: " + get);
         i++;
       }
       if (args[i].equals("-expr")) {
@@ -284,7 +287,8 @@ public int run(String[] args) throws Exception {
       if (get != null) {
         getHostDbRecord(new Path(args[0], "current"), get);
       } else {
-        readHostDb(new Path(args[0]), new Path(args[1]), dumpHomepages, dumpHostnames, expr);
+        readHostDb(new Path(args[0]), new Path(args[1]), dumpHomepages,
+            dumpHostnames, expr);
       }
       return 0;
     } catch (Exception e) {

src/java/org/apache/nutch/hostdb/UpdateHostDb.java

@@ -52,7 +52,6 @@
 import org.apache.nutch.util.NutchConfiguration;
 import org.apache.nutch.util.NutchJob;
 import org.apache.nutch.util.TimingUtil;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -78,6 +77,7 @@ public class UpdateHostDb extends Configured implements Tool {
   public static final String HOSTDB_NUMERIC_FIELDS = "hostdb.numeric.fields";
   public static final String HOSTDB_STRING_FIELDS = "hostdb.string.fields";
   public static final String HOSTDB_PERCENTILES = "hostdb.percentiles";
+  public static final String HOSTDB_UPDATEDB_DELTA_EXPRESSION = "hostdb.deltaExpression";
 
   private void updateHostDb(Path hostDb, Path crawlDb, Path topHosts,
     boolean checkFailed, boolean checkNew, boolean checkKnown,

src/java/org/apache/nutch/hostdb/UpdateHostDbMapper.java

@@ -19,15 +19,14 @@
 import java.io.IOException;
 import java.lang.invoke.MethodHandles;
 
-
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.io.FloatWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
 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.nutch.crawl.CrawlDatum;
 import org.apache.nutch.crawl.CrawlDb;
 import org.apache.nutch.crawl.NutchWritable;
@@ -36,7 +35,6 @@
 import org.apache.nutch.net.URLNormalizers;
 import org.apache.nutch.protocol.ProtocolStatus;
 import org.apache.nutch.util.URLUtil;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -62,6 +60,7 @@ public class UpdateHostDbMapper
   protected URLFilters filters = null;
   protected URLNormalizers normalizers = null;
 
+  private boolean isDeltaStatisticCalculated = false;
   public void close() {}
 
   /**
@@ -71,7 +70,8 @@ public void configure(JobConf job) {
     readingCrawlDb = job.getBoolean("hostdb.reading.crawldb", false);
     filter = job.getBoolean(UpdateHostDb.HOSTDB_URL_FILTERING, false);
     normalize = job.getBoolean(UpdateHostDb.HOSTDB_URL_NORMALIZING, false);
-
+    isDeltaStatisticCalculated = !StringUtils.isEmpty(job.get(UpdateHostDb.HOSTDB_UPDATEDB_DELTA_EXPRESSION)); 
+
     if (filter)
       filters = new URLFilters(job);
     if (normalize)
@@ -212,9 +212,10 @@ public void map(Text key, Writable value,
 
       // If we're also reading CrawlDb entries, reset db_* statistics because
       // we're aggregating them from CrawlDB anyway
-      if (readingCrawlDb) {
+
+     if (readingCrawlDb && !isDeltaStatisticCalculated) {
         hostDatum.resetStatistics();
-      }
+      } 
 
       output.collect(key, new NutchWritable(hostDatum));
     }