Salmon Run

Nutch/GORA - Delta Indexing

2012-01-29T09:24:00.000-08:00

Background

Delta, or incremental indexing, is pretty much the main reason I am so excited about Nutch/GORA. Replacing segment files with database opens up the possibility of external programs modifying metadata in the database, and thereby modifying the pipeline's behavior to support delta indexing.

For web crawling (Nutch's default use case), the Adaptive Fetch Schedule is an excellent choice. This component (set by db.fetch.schedule.class) will decrease or increase the fetch interval based on whether the page has changed or not. However, if you wanted to use Nutch as an indexing pipeline for a Content Management System (CMS) for example, then you will need to offer the CMS folks a slightly more deterministic way to control what appears in the index.

When a CMS publishes a piece of content, the content should (within some predictable time interval) make it into the index (so it starts appearing in search results). On the other hand, when a CMS unpublishes a piece of content, it should disappear from the index. In this post, I propose an approach that does this, using Nutch/GORA (with custom plugins) and some supporting infrastructure.

Infrastructure

I have previously described using a HTTP server to stage content in a local filesystem to Nutch. I use a similar strategy here, except that my seed file for a given provider (or CMS) is now a dynamically generated "index" file, also served by the HTTP server, that lists all the content available from the provider. Something like this:

In line with this, my script that launches a CherryPy HTTP server now has an additional handler to dynamically generate the index page by recursively scanning the directory (/provider_index), as well as one that serves a named file from disk (/provider). Heres the code:

#!/usr/bin/python
import cherrypy
import os
import os.path
import urllib

from cherrypy.lib.static import serve_file

SITES_DIR = "/path/to/your/sites"
SERVER_HOST = "localhost"
SERVER_PORT = 8080

def _accumulate_files(files, dirname, fnames):
  """
  This function gets called for every file (and directory) that is walked
  by os.path.walk. It accumulates the file names found into a flat array.
  The file names accumulated are relative to the providers directory.
  """
  for fname in fnames:
    abspath = os.path.join(dirname, fname)
    if os.path.isfile(abspath):
      abspath = abspath.replace(os.path.join(SITES_DIR, "providers"), "")[1:]
      files.append(abspath)

class Root:

  @cherrypy.expose
  def test(self, name):
    """
    Expose the mock site for testing.
    """
    return serve_file(os.path.join(SITES_DIR, "test", "%s.html" % (name)), \
      content_type="text/html")

  @cherrypy.expose
  def provider_index(self, name):
    """
    Builds an index page of links to all the files for the specified
    provider. The files are stored under sites/providers/$name. The
    function will recursively walk the filesystem under this directory
    and dynamically generate a flat list of links. Path separators in
    the filename are converted to "__" in the URL. The index page can
    be used as the seed URL for this content.
    """
    files = []
    os.path.walk(os.path.join(SITES_DIR, "providers", name), \
      _accumulate_files, files)
    index = "<html><head></head><body><ul>"
    for file in files:
      url = "http://%s:%s/provider/%s" % (SERVER_HOST, SERVER_PORT, \
        urllib.quote_plus(file.replace(os.path.sep, "__")))
      index += """
        <li><a href="%s">%s</a></li>
      """ % (url, url)
    index += "</ul></body></html>"
    return [index]

  @cherrypy.expose
  def provider(self, name):
    """
    Returns the contents of the XML file stored at the location 
    corresponding to the URL provided. The "__" in the URL are converted
    back to file path separators.
    """
    ct = None
    if name.endswith(".xml"):
      ct = "application/xml"
    elif name.endswith(".json"):
      ct = "application/json"
    if ct is None:
      return serve_file(os.path.join(SITES_DIR, "providers", \
        "%s" % name.replace("__", os.path.sep)), \
        content_type = "text/html")
    else:
      return serve_file(os.path.join(SITES_DIR, "providers", \
        "%s" % (urllib.unquote_plus(name).replace("__", os.path.sep))), \
        content_type = ct)

if __name__ == '__main__':
  current_dir = os.path.dirname(os.path.abspath(__file__))
  # Set up site-wide config first so we get a log if errors occur.
  cherrypy.config.update({'environment': 'production',
    'log.access_file': 'site.log',
    'log.screen': True,
    "server.socket_host" : SERVER_HOST,
    "server.socket_port" : SERVER_PORT})
  cherrypy.quickstart(Root(), '/')

My seed file now lists only the URL to the index file, as shown below:

1	http://localhost:8080/provider_index/prov1 u_idx=prov1

Delta Indexing Overview

As I mentioned earlier, the CMS can send either a publish or an unpublish request. A publish request can translate to an Add (if its a new page that doesn't exist in the database) or an Update. An unpublish request would translate to a Delete (or Hide) request at the database level.

For an Add request, we reset the fetch time of the provider page to the current time minus the fetch interval, making it eligible for a fetch in the next crawl run. Since it is dynamically generated off the file system, the added file is guaranteed to be in this file (since it exists in the filesystem). In the first iteration of the recrawl, the index page will be refetched and reparsed, and the second iteration will generate the newly added page(s) from the index page outlinks to the fetch list, and then to the index.

Of course, the index page itself should not make it into the index, so we have a custom Index Filter (described later) to filter these out by pattern.

Here is an alternative approach which relies on accurate Last-Modified headers and a more aggressive crawl schedule. With this approach, you don't need an index file or any custom handling for Adds and Updates, but you do need to crawl more frequently. I prefer an event based approach, which is what I have here.

For an Update request, we reset the fetch time of the page specified by the URL, similar to the index page. The first iteration of the next recrawl will pick the changes up and update it into the index.

For a Delete, we do two things - first, we delete the record from Solr. Then we mark the record deleted in Cassandra by setting the status to STATUS_GONE. We add logic to an indexing filter to filter out pages with this status from making it into the index.

Delta Indexing Tool

To do the publish and unpublish on the index, I built a tool (callable using full class name from bin/nutch). For this I built my package under nutch's src/java tree. I had to modify the nutch build.xml slightly to get my code to compile, specifically add the com/mycompany/nutch/**/*.java path to the javac includes in the "compile-core" target of nutch's build.xml.

I had initially planned on building it with GORA so it could be easily translated to other backends as well. The GORA tutorial has examples of retrieving and updating records by key. However, it turns out that the CassandraStore (the gora-cassandra implementation of the DataStore), has an empty implementation of get(String) that returns null.

So I finally settled on using the Hector API (which GORA also uses) to talk directly with the Cassandra database. I am using Hector's template API which feels similar to (and is probably inspired by) Spring's JdbcTemplate. Here is the code for this tool. Not as nice as using GORA (ie not database agnostic), but it'll have to do for now.

// Source: src/java/com/mycompany/nutch/tools/DeltaHandler.java
package com.mycompany.nutch.tools;

import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.CassandraHostConfigurator;
import me.prettyprint.cassandra.service.template.ColumnFamilyResult;
import me.prettyprint.cassandra.service.template.ColumnFamilyTemplate;
import me.prettyprint.cassandra.service.template.ColumnFamilyUpdater;
import me.prettyprint.cassandra.service.template.ThriftColumnFamilyTemplate;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.factory.HFactory;

import org.apache.avro.util.Utf8;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.nutch.crawl.CrawlStatus;
import org.apache.nutch.util.TableUtil;
import org.apache.solr.client.solrj.impl.CommonsHttpSolrServer;

public class DeltaHandler {
  
  private static final Log LOG = LogFactory.getLog(DeltaHandler.class);
  
  private static final String PROVIDER_INDEX_URLPREFIX_KEY =
    "mycompany.provider.index.urlprefix";
  private static final String SOLR_URL = "mycompany.solr.url";
  private static final Utf8 U_IDX = new Utf8("u_idx");
  
  private String providerIndexUrlPrefix;
  private CommonsHttpSolrServer solrServer;
  private Keyspace keyspace;
  private ColumnFamilyTemplate<String,String> template;
  
  public DeltaHandler() {
    try {
      init();
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }
  
  private void init() throws Exception {
    Configuration conf = new Configuration();
    conf.addResource("nutch-default.xml");
    conf.addResource("nutch-site.xml");
    this.providerIndexUrlPrefix = conf.get(PROVIDER_INDEX_URLPREFIX_KEY);
    this.solrServer = new CommonsHttpSolrServer(conf.get(SOLR_URL));
    Cluster cluster = HFactory.getOrCreateCluster("Test Cluster", 
      new CassandraHostConfigurator("localhost:9160"));
    this.keyspace = HFactory.createKeyspace("webpage", cluster);
    this.template = new ThriftColumnFamilyTemplate<String,String>(
      keyspace, "f", StringSerializer.get(), StringSerializer.get()
    );
  }
  
  private void destroy() {
  }
  
  public void publish(String url, String idx) throws Exception {
    String key = TableUtil.reverseUrl(url);
    ColumnFamilyResult<String,String> res = 
      template.queryColumns(key);
    String bas = res.getString("bas"); 
    if (StringUtils.isEmpty(bas)) {
      // requested URL does not exist, should be an "add",
      // reset fetchtime for the index page
      key = TableUtil.reverseUrl(StringUtils.join(new String[] {
        providerIndexUrlPrefix, idx}, "/"));
      res = template.queryColumns(key);
      bas = res.getString("bas");
    }
    if (StringUtils.isEmpty(bas)) return;
    int fetchInterval = Integer.valueOf(res.getString("fi"));
    // update the fetch time to current - fetchInterval so
    // it is eligible for crawling immediately
    ColumnFamilyUpdater<String,String> updater =
      template.createUpdater(key);
    updater.setString("ts", String.valueOf(
      System.currentTimeMillis() - fetchInterval));
    template.update(updater);
  }
  
  public void unpublish(String url, String idx, boolean commit) 
      throws Exception {
    String key = TableUtil.reverseUrl(url);
    ColumnFamilyResult<String,String> res = template.queryColumns(key);
    String bas = res.getString("bas");
    if (StringUtils.isNotEmpty(bas)) {
      System.out.println("found it!");
      ColumnFamilyUpdater<String,String> updater = 
        template.createUpdater(key);
      updater.setString("st", String.valueOf(
        CrawlStatus.STATUS_GONE));
      template.update(updater);
      deleteFromSolr(key, commit);
    }
  }
  
  private void deleteFromSolr(String key, boolean commit) 
      throws Exception {
    solrServer.deleteById(key);
    if (commit) {
      solrServer.commit();
    }
  }
  
  private static void usage() {
    System.out.println(
      "Usage: DeltaHandler publish|unpublish url idx [commit]");
    System.out.println("commit = true|false, only for unpublish");
    System.exit(-1);
  }
  
  public static void main(String[] args) {
    String command = null;
    String url = null;
    String idx = null;
    Boolean commit = null;
    if (args.length > 0) {
      command = args[0];
      if (!("publish".equals(command)) && 
          !("unpublish".equals(command))) {
        usage();
      }
    }
    if ("publish".equals(command)) {
      if (args.length > 2) {
        url = args[1];
        idx = args[2];
      } else {
        usage();
      }
    } else if ("unpublish".equals(command)) {
      if (args.length > 3) {
        url = args[1];
        idx = args[2];
        commit = Boolean.valueOf(args[3]);
      } else {
        usage();
      }
    }
    DeltaHandler handler = null;
    try {
      handler = new DeltaHandler();
      if ("publish".equals(command)) {
        handler.publish(url, idx);
      } else {
        handler.unpublish(url, idx, commit);
      }
    } catch (Exception e) {
      LOG.error(e, e);
    } finally {
      if (handler != null) {
        handler.destroy();
      }
    }
  }
}

This is really only for testing. For high volume use, it would be fairly simple to have this logic live behind a webservice which the CMS code could call. But in any case, here are examples of command line usage of the tool above.

sujit@cyclone:local$ # publishing a page
sujit@cyclone:local$ bin/nutch com.mycompany.nutch.tools.DeltaHandler \
  publish http://localhost:8080/provider/prov1__1__000022.xml prov1
sujit@cyclone:local$ # unpublishing a page
sujit@cyclone:local$ bin/nutch com.mycompany.nutch.tools.DeltaHandler \
  unpublish http://localhost:8080/provider/prov1__1__000022.xml prov1 \
  true

I also added a few new properties for this tool (and for the plugin decribed below). Here are the additional properties from my nutch-site.xml file.

<property>
  <name>mycompany.provider.index.urlprefix</name>
  <value>http://localhost:8080/provider_index</value>
  <description>The URL to the content server hosting the provider content  </description>
</property>

<property>
  <name>mycompany.solr.url</name>
  <value>http://localhost:8983/solr</value>
  <description>The SOLR URL to publish/unpublisth to</description>
</property>

<property>
  <name>mycompany.provider.index.pattern</name>
  <value>provider_index</value>
  <description>Pattern for "meta" pages listing other local pages. This
  page is needed for delta indexing to manage on-demand add/delete/update
  of pages from collections which need this feature. But it should not
  be indexed, so need to be removed during the indexing step.</description>
</property>

Delta Indexing Filter Plugin

As mentioned above, you probably don't want your index pages to make it into the index, since they are basically link farms and have no useful content (for the search user). So you want to suppress these pages from ever making it into the index.

The second class of pages would be the ones marked GONE by the unpublish command. The unpublish deletes the record from the Solr index, but you want to make sure that the page doesn't slip in on the next solrindex call. So we build an indexing filter which filters out these two category of pages.

The functionality above is built into the DeltaIndexFilter, a simple Nutch IndexFilter implementation. Here is the declaration for this filter from my plugin.xml file.

  <extension id="com.mycompany.nutch.indexer.provider"
      name="Delta Indexing related Page Index Suppressor"
      point="org.apache.nutch.indexer.IndexingFilter">
    <implementation id="mycompany-indexer-provider"
        class="com.mycompany.nutch.indexer.delta.DeltaIndexingFilter"/>
  </extension>

and here is the code for the DeltaIndexFilter.

// Source: src/plugin/mycompany/src/java/com/mycompany/nutch/indexer/delta/DeltaIndexingFilter.jav
package com.mycompany.nutch.indexer.delta;

import java.util.Collection;
import java.util.HashSet;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.nutch.crawl.CrawlStatus;
import org.apache.nutch.indexer.IndexingException;
import org.apache.nutch.indexer.IndexingFilter;
import org.apache.nutch.indexer.NutchDocument;
import org.apache.nutch.storage.WebPage;
import org.apache.nutch.storage.WebPage.Field;

public class DeltaIndexingFilter implements IndexingFilter {

  private static final String PROVIDER_INDEX_PATTERN_KEY = 
    "mycompany.provider.index.pattern";
  
  private static final Set<WebPage.Field> FIELDS = 
    new HashSet<WebPage.Field>();
  static {
    FIELDS.add(WebPage.Field.STATUS);
    FIELDS.add(WebPage.Field.METADATA);
  }
  
  private Configuration conf;
  private String providerIndexPattern;
  
  @Override
  public NutchDocument filter(NutchDocument doc, String url, WebPage page)
      throws IndexingException {
    if (StringUtils.isEmpty(providerIndexPattern)) {
      return doc;
    } else {
      if (url.contains(providerIndexPattern) || 
          CrawlStatus.STATUS_GONE == page.getStatus()) {
        // do not index this page
        return null;
      } else {
        return doc;
      }
    }
  }

  @Override
  public Collection<Field> getFields() {
    return FIELDS;
  }

  @Override
  public Configuration getConf() {
    return conf;
  }

  @Override
  public void setConf(Configuration conf) {
    this.conf = conf;
    this.providerIndexPattern = conf.get(
      PROVIDER_INDEX_PATTERN_KEY);
  }
}

And thats pretty much it! Delta indexing in just a few lines (relatively speaking of course - this is Java we are talking about :-)), thanks to Nutch/GORA (which gives us the Cassandra database) and Hector (which gives us the API to write to it from outside Nutch).

Nutch/GORA - Parsing Custom XML

2012-01-20T22:04:00.000-08:00

This week, I continue my exploration of Nutch/GORA to see if I can use it to parse and index custom (non-web) XML content. In addition to crawling selected medical sites for our search index, we also rely on content providers who provide us encyclopedic medical content in different areas. These providers periodically provide us data, each in their own custom XML format, which we parse and ingest into our indexes. In this post, I describe how I plan to do this using Nutch/GORA.

Nutch provides the XMLParser_Plugin as an add-on. It can be configured to parse a particular format using XPath expressions. I looked at it briefly, but I abandoned it because it wasn't clear how it would be able to parse multiple XML formats (one of my requirements).

Of course, one could settle on a common XML format as proposed in this proposal and then use this plugin. That is certainly an option, and it allows the parsing work to be farmed out - XML is almost like a second language for Java application programmers, unlike the backend types who do the crawling and indexing stuff. But the problem with this approach is the extra step, and the fact that the fetched content is no longer the original content we got.

The approach I came up with was to write a Nutch parse plugin that uses the u_idx metadata field (described in my previous post) to decide which parser to spin up for a given XML page. The XML parsers all take a String representing the fetched content and return a map of the parsed fields. This still allows the division of labor thing I spoke of earlier - the parsers are simple XML parsers, and the plugin deals with the details of updating the WebPage data back into Cassandra.

Exposing Content over HTTP

The first step is to expose the provider XML content over an HTTP interface. Right now I am developing/testing in local mode, but at some point I plan on running this in distributed mode, and having the content available over HTTP just makes more sense. So I added another handler to my CherryPy server to serve XML files with content type "application/xml". The updated code is shown below:

#!/usr/bin/python
import os.path
import cherrypy
from cherrypy.lib.static import serve_file

SITES_DIR="/path/to/my/data/directory"

class Root:

  @cherrypy.expose
  def test(self, name):
    return serve_file(os.path.join(SITES_DIR, "test", "%s.html" % (name)), \
      content_type="text/html")

  @cherrypy.expose
  def providers(self, name):
    return serve_file(os.path.join(SITES_DIR, "providers", \
      "%s" % (name.replace("__", "/"))), content_type="application/xml")

if __name__ == '__main__':
  current_dir = os.path.dirname(os.path.abspath(__file__))
  # Set up site-wide config first so we get a log if errors occur.
  cherrypy.config.update({'environment': 'production',
    'log.access_file': 'site.log',
    'log.screen': True,
    "server.socket_host" : "127.0.0.1",
    "server.socket_port" : 8080})
  cherrypy.quickstart(Root(), '/')

And my seed URL (generated by running a find command followed by a few text transformations in vim) looks like this:

http://localhost:8080/providers/prov1__1__000001.xml  u_idx=prov1
http://localhost:8080/providers/prov1__1__000002.xml  u_idx=prov1
http://localhost:8080/providers/prov1__1__000003.xml  u_idx=prov1
http://localhost:8080/providers/prov1__1__000004.xml  u_idx=prov1
http://localhost:8080/providers/prov1__1__000005.xml  u_idx=prov1
http://localhost:8080/providers/prov1__1__000006.xml  u_idx=prov1
http://localhost:8080/providers/prov1__1__000007.xml  u_idx=prov1
http://localhost:8080/providers/prov1__1__000008.xml  u_idx=prov1
http://localhost:8080/providers/prov1__1__000010.xml  u_idx=prov1
...

The XML files themselves are provided in a multi-directory format - path separators in the file name (relative to the root of the local site) become "__" in the URLs, and the handler makes the transformation to read and serve the appropriate file.

Parse Plugin Code

To enable the custom parsing, I added a parse plugin to my "mycompany" plugin project that uses the u_idx value to parse the content with the appropriate XML processor. I decided to use JDOM as my parsing library since this is already provided in the nutch distribution. Here is my updated plugin.xml file, the definition for the parse plugin is in the last block.

<?xml version="1.0" encoding="UTF-8"?>
<!-- Source: src/plugin/mycompany/plugins.xml -->
<plugin id="mycompany" 
    name="Custom Company-specific plugins"
    version="1.0.0"
    provider-name="My Company">

  <runtime>
    <library name="mycompany.jar">
      <export name="*"/>
    </library>
    <library name="jdom-1.1.jar"/>
  </runtime>

  <requires>
    <import plugin="nutch-extensionpoints"/>
  </requires>

  <extension id="com.mycompany.nutch.indexer.usertags"
      name="User Tag Indexing Filter"
      point="org.apache.nutch.indexer.IndexingFilter">
    <implementation id="mycompany-indexer-usertags"
        class="com.mycompany.nutch.indexer.usertags.UserTagIndexingFilter"/>
  </extension>

  <extension id="com.mycompany.nutch.scoring.usertags"
      name="User Tag Scoring Filter"
      point="org.apache.nutch.scoring.ScoringFilter">
    <implementation id="mycompany-scoring-metadata"
        class="com.mycompany.nutch.scoring.usertags.UserTagScoringFilter"/>
  </extension>

  <extension id="com.mycompany.nutch.parse.xml"
      name="Provider XML Parser Plugin"
      point="org.apache.nutch.parse.Parser">
    <implementation id="mycompany-parse-provider-xml"
        class="com.mycompany.nutch.parse.xml.ProviderXmlParser">
      <parameter name="contentType" value="application/xml"/>
      <parameter name="pathSuffix" value="xml"/>
    </implementation>
  </extension>

</plugin>

The Java code for the parse plugin is shown below:

// Source: src/plugin/mycompany/src/java/com/mycompany/nutch/parse/xml/ProviderXmlParser.java
package com.mycompany.nutch.parse.xml;

import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.avro.util.Utf8;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.nutch.parse.Outlink;
import org.apache.nutch.parse.Parse;
import org.apache.nutch.parse.ParseStatusCodes;
import org.apache.nutch.parse.Parser;
import org.apache.nutch.storage.ParseStatus;
import org.apache.nutch.storage.WebPage;
import org.apache.nutch.storage.WebPage.Field;
import org.apache.nutch.util.Bytes;

public class ProviderXmlParser implements Parser {

  private static final Log LOG = LogFactory.getLog(ProviderXmlParser.class); 
  private static final Set<WebPage.Field> FIELDS = new HashSet<WebPage.Field>();
  private static final Utf8 IDX_KEY = new Utf8("u_idx");
  
  static {
    FIELDS.add(WebPage.Field.METADATA);
    FIELDS.add(WebPage.Field.OUTLINKS);
  }

  private Configuration conf;
  
  @Override
  public Parse getParse(String url, WebPage page) {
    Parse parse = new Parse();
    parse.setParseStatus(new ParseStatus());
    parse.setOutlinks(new Outlink[0]);
    Map<Utf8,ByteBuffer> metadata = page.getMetadata();
    if (metadata.containsKey(IDX_KEY)) {
      String idx = Bytes.toString(Bytes.toBytes(metadata.get(IDX_KEY)));
      IProviderXmlProcessor processor = 
        ProviderXmlProcessorFactory.getProcessor(idx);
      if (processor != null) {
        try {
          LOG.info("Parsing URL:[" + url + "] with " + 
            processor.getClass().getSimpleName());
          Map<String,String> parsedFields = processor.parse(
              Bytes.toString(Bytes.toBytes(page.getContent())));
          parse.setText(parsedFields.get(ProviderXmlFields.content.name()));
          parse.setTitle(parsedFields.get(ProviderXmlFields.title.name()));
          // set the rest of the metadata back into the page
          for (String key : parsedFields.keySet()) {
            if (ProviderXmlFields.content.name().equals(key) ||
                ProviderXmlFields.title.name().equals(key)) {
              continue;
            }
            page.putToMetadata(new Utf8(key), 
              ByteBuffer.wrap(parsedFields.get(key).getBytes()));
          }
          parse.getParseStatus().setMajorCode(ParseStatusCodes.SUCCESS);
        } catch (Exception e) {
          LOG.warn("Parse of URL: " + url + " failed", e);
          LOG.warn("content=[" + Bytes.toString(Bytes.toBytes(
            page.getContent())) + "]");
          parse.getParseStatus().setMajorCode(ParseStatusCodes.FAILED);
        }
      }
    }
    return parse;
  }

  @Override
  public Collection<Field> getFields() {
    return FIELDS;
  }

  @Override
  public Configuration getConf() {
    return conf;
  }

  @Override
  public void setConf(Configuration conf) {
    this.conf = conf;
  }
}

As you can see, the getParse() method reads the u_idx metadata from the page, and based on that asks the ProviderXmlProcessorFactory for the appropriate processor. The factory returns the appropriate IProviderXmlProcessor implementation for the content type (or a null if no implementation can be found). The IProviderXmlProcessor takes the content as a string, parses it with custom logic and returns a Map of metadata names and values. The plugin than populates the Parse object and directly updates the WebPage with the metadata. Here is the rest of the code, without too much explanation.

// Source: src/plugin/mycompany/src/java/com/mycompany/nutch/parse/xml/ProviderXmlProcessorFactory.java
package com.mycompany.nutch.parse.xml;

import java.util.HashMap;
import java.util.Map;

public class ProviderXmlProcessorFactory {

  private static ProviderXmlProcessorFactory FACTORY = 
    new ProviderXmlProcessorFactory();
  private Map<String,IProviderXmlProcessor> processors = 
    new HashMap<String,IProviderXmlProcessor>();
  
  private ProviderXmlProcessorFactory() {
    processors.put("prov1", new Prov1XmlProcessor());
    // no more for now
  }
  
  public static IProviderXmlProcessor getProcessor(String idx) {
    return FACTORY.processors.get(idx);
  }
}

// Source: src/plugin/mycompany/src/java/com/mycompany/nutch/parse/xml/IProviderXmlProcessor.java
package com.mycompany.nutch.parse.xml;

import java.util.Map;

public interface IProviderXmlProcessor {

  public Map<String,String> parse(String content) throws Exception;

}

// Source: src/plugin/mycompany/src/java/com/mycompany/nutch/parse/xml/Prov1XmlProcessor.java
package com.mycompany.nutch.parse.xml;

import java.io.ByteArrayInputStream;
import java.util.HashMap;
import java.util.Map;

import org.jdom.Document;
import org.jdom.Element;
import org.jdom.input.SAXBuilder;
import org.xml.sax.InputSource;

public class Prov1XmlProcessor implements IProviderXmlProcessor {

  @Override
  public Map<String,String> parse(String content) throws Exception {
    Map<String,String> parsedFields = new HashMap<String,String>();
    SAXBuilder builder = new SAXBuilder();
    Document doc = builder.build(new InputSource(
      new ByteArrayInputStream(content.getBytes())));
    Element root = doc.getRootElement();
    parsedFields.put(ProviderXmlFields.u_lang.name(), 
      root.getAttributeValue("language"));
    parsedFields.put(ProviderXmlFields.title.name(), 
      root.getAttributeValue("title"));
    parsedFields.put(ProviderXmlFields.u_category.name(), 
      root.getAttributeValue("subContent"));
    parsedFields.put(ProviderXmlFields.u_contentid.name(), 
      root.getAttributeValue("genContentID"));
    Element versionInfo = root.getChild("versionInfo");
    if (versionInfo != null) {
      parsedFields.put(ProviderXmlFields.u_reviewdate.name(), 
        ProviderXmlParserUtils.convertToIso8601(
        versionInfo.getAttributeValue("reviewDate")));
      parsedFields.put(ProviderXmlFields.u_reviewers.name(), 
        versionInfo.getAttributeValue("reviewedBy"));
    }
    parsedFields.put(ProviderXmlFields.content.name(), 
      ProviderXmlParserUtils.getTextContent(root));
    return parsedFields;
  }
}

In addition, I built some standard functions and put them in their own utility class, and an enum which lists the fields available.

// Source: /src/plugin/mycompany/src/java/com/mycompany/nutch/parse/xml/ProviderXmlParserUtils.java
package com.mycompany.nutch.parse.xml;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

import org.apache.commons.lang.time.DateUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jdom.Element;

public class ProviderXmlParserUtils {

  private static final Log LOG = LogFactory.getLog(ProviderXmlParserUtils.class);
  
  /////////// normalize incoming dates to ISO-8601 ///////////
  
  private static final String[] DATE_PATTERNS = new String[] {
    "yyyyMMdd",
    "MM/dd/yyyy",
  };
  private static final SimpleDateFormat ISO_8601_DATE_FORMAT =
    new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS");
  
  public static String convertToIso8601(String date) {
    try {
      Date d = DateUtils.parseDate(date, DATE_PATTERNS);
      return ISO_8601_DATE_FORMAT.format(d);
    } catch (Exception e) {
      LOG.warn("Cannot convert date: " + date + " to ISO-8601 format", e);
      return "";
    }
  }

  ///////////// get text content of XML //////////////////
  
  public static String getTextContent(Element root) {
    StringBuilder buf = new StringBuilder();
    getTextContent_r(buf, root);
    return buf.toString();
  }
  
  @SuppressWarnings("unchecked")
  private static void getTextContent_r(StringBuilder buf, Element e) {
    buf.append(e.getTextTrim());
    List<Element> children = e.getChildren();
    for (Element child : children) {
      getTextContent_r(buf, child);
    }
    buf.append(" ");
  }
}

// Source: src/plugin/mycompany/src/java/com/mycompany/nutch/parse/xml/ProviderXmlFields.java
package com.mycompany.nutch.parse.xml;

public enum ProviderXmlFields {

  // common
  title,
  content,
  // prov1 
  u_lang,
  u_category,
  u_contentid,
  u_reviewdate,
  u_reviewers,
}

Note that while I have put all the classes in the same package for convenience, there is nothing preventing someone (and is probably recommended in an environment where you want to move the XML parsing work to a different group) from putting everything other than the actual plugin into a separate project and declaring the JAR from that project as a dependency.

To test it, I repeatedly dropped the webpage from Cassandra, ran inject, generate, fetch and parse and looked at the Cassandra records usng my display_webpage.py script and at error messages in logs/hadoop.log.

The first time I ran this code (well, not the first time, but you get the idea), I got the following exception in the logs/hadoop.log.

2012-01-18 15:28:00,860 INFO  parse.ParserFactory - The parsing plugins: 
[org.apache.nutch.parse.tika.TikaParser] are enabled via the plugin.includes 
system property, and all claim to support the content type application/xml, 
but they are not mapped to it  in the parse-plugins.xml file
2012-01-18 15:28:01,219 ERROR tika.TikaParser - Error parsing http://localhost:
8080/providers/prov1__1__000008.xmlorg.apache.tika.exception.TikaException: XML
parse error at org.apache.tika.parser.xml.XMLParser.parse(XMLParser.java:71)
        at org.apache.nutch.parse.tika.TikaParser.getParse(TikaParser.java:117)
        at org.apache.nutch.parse.ParseCallable.call(ParseCallable.java:36)
        at org.apache.nutch.parse.ParseCallable.call(ParseCallable.java:23)
        at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
        at java.util.concurrent.FutureTask.run(FutureTask.java:138)
        at java.lang.Thread.run(Thread.java:680)

This told me that the (preconfigured) Tika XML parser was getting to my data first, which I did not want, so I disabled it in the nutch-site.xml plugin.includes property. My plugin.includes now looks like this:

<!-- From this:
<property>
  <name>plugin.includes</name>
  <value>protocol-http|urlfilter-regex|parse-(html|tika)|\
         index-(basic|anchor)|urlnormalizer-(pass|regex|\
         basic)|scoring-opic|mycompany</value>
  <description/>
</property>

to this: -->

<property>
  <name>plugin.includes</name>
  <value>protocol-http|urlfilter-regex|parse-html|\
         index-(basic|anchor)|urlnormalizer-(pass|regex|\
         basic)|scoring-opic|mycompany</value>
  <description/>
</property>

Indexing

After this, a couple of iterations later, I got a clean parse run. I followed that up with updatedb, then modified my old UserTagIndexingFilter to pull all metadata columns whose key is prefixed with u_ (meaning user-defined). Here is the updated code for the UserTagIndexingFilter.java.

// Source: src/plugin/mycompany/src/java/com/mycompany/nutch/indexer/usertags/UserTagIndexingFilter.java
package com.mycompany.nutch.indexer.usertags;

import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.avro.util.Utf8;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.nutch.indexer.IndexingException;
import org.apache.nutch.indexer.IndexingFilter;
import org.apache.nutch.indexer.NutchDocument;
import org.apache.nutch.storage.WebPage;
import org.apache.nutch.storage.WebPage.Field;
import org.apache.nutch.util.Bytes;
import org.apache.nutch.util.TableUtil;

public class UserTagIndexingFilter implements IndexingFilter {

  private static final Log LOG = LogFactory.getLog(UserTagIndexingFilter.class);
  private static final String INHERITABLE_USERTAGS_LIST_PROP = 
    "mycompany.usertags.inheritable";
  
  private static final Set<WebPage.Field> FIELDS = new HashSet<WebPage.Field>();
  
  static {
    FIELDS.add(WebPage.Field.METADATA);
  }
  
  private Configuration conf;
  private Set<String> inheritableUserTags;
  
  @Override
  public NutchDocument filter(NutchDocument doc, String url, 
      WebPage page) throws IndexingException {
    LOG.info("Adding user tags for page:" + url);
    Map<Utf8,ByteBuffer> metadata = page.getMetadata();
    for (Utf8 key : metadata.keySet()) {
      String keyStr = TableUtil.toString(key);
      if (StringUtils.isEmpty(keyStr)) {
        continue;
      }
      if (keyStr.startsWith("u_")) {
        String value = Bytes.toString(
          Bytes.toBytes(metadata.get(key)));
        doc.add(keyStr, value);
      }
    }
    return doc;
  }

  @Override
  public Configuration getConf() {
    return conf;
  }

  @Override
  public void setConf(Configuration conf) {
    this.conf = conf;
  }

  @Override
  public Collection<Field> getFields() {
    return FIELDS;
  }
}

I also needed to add the new fields I was pulling out to the conf/solrindex-mapping.xml:

<?xml version="1.0" encoding="UTF-8"?>
<!-- conf/solrindex-mapping.xml -->
<mapping>
  ...
  <fields>
    <field dest="content" source="content"/>
    <!-- more nutch defined fields here -->
    ...
    <!-- user defined -->
    <field dest="u_idx" source="u_idx"/>
    <field dest="u_contentid" source="u_contentid"/>
    <field dest="u_category" source="u_category"/>
    <field dest="u_lang" source="u_lang"/>
    <field dest="u_reviewdate" source="u_reviewdate"/>
    <field dest="u_reviewers" source="u_reviewers"/>
  </fields>
  <uniqueKey>id</uniqueKey>
</mapping>

and to the Solr schema.xml file (followed by a Solr restart).

<?xml version="1.0" encoding="UTF-8" ?>
<!-- Source: conf/schema.xml (copied to solr's example/solr/conf) -->
<schema name="nutch" version="1.4">
  ...
        <field name="u_idx" type="string" stored="true" indexed="true"/>
        <field name="u_contentid" type="string" stored="true" indexed="true"/>
        <field name="u_category" type="string" stored="true" indexed="true"/>
        <field name="u_lang" type="string" stored="true" indexed="true"/>
        <field name="u_reviewdate" type="string" stored="true" indexed="false"/>
        <field name="u_reviewers" type="string" stored="true" indexed="false"/>
  ...
</schema>

Running the nutch solrindex job now promotes all the XML files (along with the parsed metadata) into the Solr index.

Conclusion

The approach described works well when an XML document corresponds to one WebPage record in Cassandra (and one Solr record). But in some cases, we parse the file up into sections and do other stuff with it. In such cases, it may be better to go with the intermediate XML markup approach described in the proposal above, or even a hybrid approach where the original file goes through the pipeline as described, then goes through an additional external parse phase which produces section files (for example) in the intermediate XML format. I haven't thought through this thing in great detail though, I plan on doing this after I am done checking out the basic stuff.

Nutch/GORA - Scoring and Indexing Plugins for Metadata Inheritance

2012-01-13T23:39:00.000-08:00

Last week, I described my initial explorations with Nutch/GORA and Cassandra to crawl a small external site. At the end of the post, I raised three issues, which, as Julien Nioche points out, are already supported via Nutch. This week, I describe a pair of Nutch Plugins to do metadata inheritance - this is based heavily on the ideas in the Nutch Writing Plugin Example Wiki Page - the example in this page does almost what I want, but works with the traditional segment files, not the Nutch/GORA based WebPage data structure. So I figured that the code may be worth sharing, for people who would like to leverage Nutch/GORA.

What is Metadata Inheritance?

What I mean by this is that I want to inject some metadata with the seed URL, and the metadata injected should be propagated to all the children of that page, recursively. I want to use the same infrastructure to crawl the web, as well as crawl internal resources, and I want some way to distinguish the various sources, so the one piece of metadata I add to the seed URL is the u_idx value.

Mocking the Target Site

Before I started, I decided to mock out my own "site" using a bunch of autogenerated HTML files served up using CherryPy. Since I was just starting out with Nutch/GORA, I expected it to be a few iterations before I got things to work correctly, and it seemed a bit unfair to slam an external site every 5-10 minutes. Plus, doing this allows me to work on the code without having to have an internet connection, which is almost all the time for me (most of the stuff I write about is done in a train with no internet connection, during my commute to and from work). In any case, my mock site looks like this:

  A.html
  |
  +-- B.html
  |   |
  |   +-- D.html
  |   |
  |   +-- E.html
  |
  +-- C.html
  |   |
  |   +-- F.html
  |   |
  |   +-- G.html
  |

The code to build this site is quite simple, just a few lines of Python code. Here it is if you are interested.

#!/usr/bin/python
import sys
import os

def gen_page(dir, prefix, pages):
  curr_page = pages[0]
  sub_pages = pages[1:]
  f = open(os.path.join(dir, curr_page + ".html"), "w")
  c = """<html>
    <head><title>Test Site Page %s</title></head>
    <body>
      <p>Random text for Page %s</p>
      <ul>
        <li><a href=\"%s%s\">Page %s</a></li>
        <li><a href=\"%s%s\">Page %s</a></li>
      </ul>
    </body>
    </html>
  """ 
  f.write(c % (curr_page, curr_page, prefix, sub_pages[0], \
    sub_pages[0], prefix, sub_pages[1], sub_pages[1]))
  f.close()

def main():
  prefix = "http://localhost:8080/test/"
  dir = "../test_site"
  # depth 0
  gen_page(dir, prefix, ["A", "B", "C"])
  # depth 1
  gen_page(dir, prefix, ["B", "D", "E"])
  gen_page(dir, prefix, ["C", "F", "G"])
  # depth 2
  gen_page(dir, prefix, ["D", "H", "I"])
  gen_page(dir, prefix, ["E", "J", "K"])
  gen_page(dir, prefix, ["F", "L", "M"])
  gen_page(dir, prefix, ["G", "N", "O"])

if __name__ == "__main__":
  main()

So is the code to serve this site locally on port 8080, using CherryPy. I could have used a standard HTTP server like ngnix or Apache HTTPD instead, but I had CherryPy already available from a previous project, and I expected that I'll need a programmable HTTP interface at some point in this project anyway, so sooner the better.

#!/usr/bin/python
import os.path
import cherrypy
from cherrypy.lib.static import serve_file

class Root:

  @cherrypy.expose
  def test(self, name):
    return serve_file(os.path.join(""/path/to/site_root", "%s.html" % \
      (name)), content_type="text/html")

if __name__ == '__main__':
  current_dir = os.path.dirname(os.path.abspath(__file__))
  # Set up site-wide config first so we get a log if errors occur.
  cherrypy.config.update({'environment': 'production',
    'log.error_file': 'site.log',
    'log.screen': True,
    "server.socket_host" : "127.0.0.1",
    "server.socket_port" : 8080})
  cherrypy.quickstart(Root(), '/')

My seed file points to the A.html file, like so:

1	http://localhost:8080/test/A u_idx=test

Project Setup

I also spent some time getting Nutch to work with Ivy inside Eclipse and on the command line. I found the instructions on this Pentaho Wiki Page very helpful. The way I got the sftp-protocol's ivy.xml configuration to work on both Eclipse and command line (pointed out by Alexis in his TechVineyard post) was to just put the full path in it - not terribly portable, but works for me.

I also moved out all the configuration overrides into conf/nutch-site.xml like I should have done to begin with, and added the extra libraries I spoke of in my previous post into nutch/lib, so calls to "ant" just repopulates the runtime/local with the updated code, libraries and configuration. Here is my nutch-site.xml file.

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Source: conf/nutch-site.xml -->

<configuration>

<property>
  <name>http.agent.name</name>
  <value>CNC_Crawler</value>
  <description/>
</property>

<property>
  <name>http.robots.agents</name>
  <value>CNC_Crawler,*</value>
  <description/>
</property>

<property>
  <name>storage.data.store.class</name>
  <value>org.apache.gora.cassandra.store.CassandraStore</value>
  <description>Default class for storing data</description>
</property>

<property>
  <name>db.ignore.external.links</name>
  <value>true</value>
  <description>If true, outlinks leading from a page to external hosts
  will be ignored. This is an effective way to limit the crawl to include
  only initially injected hosts, without creating complex URLFilters.
  </description>
</property>

<property>
  <name>mycompany.usertags.inheritable</name>
  <value>u_idx</value>
  <description>Comma-separated list of user tags which should be inherited
  by child pages.
  </description>
</property>

<property>
  <name>plugin.includes</name>
 <value>protocol-http|urlfilter-regex|parse-(html|tika)|\
        index-(basic|anchor)|urlnormalizer-(pass|regex|\
        basic)|scoring-opic|mycompany</value>
 <description>Regular expression naming plugin directory names to
  include.  Any plugin not matching this expression is excluded.
  In any case you need at least include the nutch-extensionpoints plugin. By
  default Nutch includes crawling just HTML and plain text via HTTP,
  and basic indexing and search plugins. In order to use HTTPS please enable
  protocol-httpclient, but be aware of possible intermittent problems with the
  underlying commons-httpclient library.
  </description>
</property>

</configuration>

I also built myself a little script that does the generate/fetch/parse/inject dance for a specified depth. Its still a bit of a work in progress - inject doesn't work at the moment, but I am working on it :-). At the end of each iteration, it prints out the counts in the "f" and "p" column families, as well as the count of records indexed to Solr. Here it is:

#!/usr/bin/python
# Takes a depth parameter and repeatedly calls inject, fetch, generate, 
# parse, update that many times

import getopt
import os
import os.path
import pycassa
import simplejson
import subprocess
import sys
import urllib

from pycassa.pool import ConnectionPool
from pycassa.util import OrderedDict
from urllib2 import *

# configuration
NUTCH_HOME = "/path/to/your/nutch/runtime/local"
SOLR_URL = "http://127.0.0.1:8983/solr/"
CASSANDRA_HOST_PORT = "localhost:9160"

def inject(depth, seed):
  print "`:::: Injecting (depth %d) from seed: %s ::::" % (depth, seed)
  cwd = os.getcwd()
  os.chdir(NUTCH_HOME)
  p = os.popen("bin/nutch inject", 'r')
  for line in p.readlines():
    print line,
  p.close()
  os.chdir(cwd)

def generate(depth):
  print ":::: Generating (depth %d) ::::" % (depth)
  cwd = os.getcwd()
  os.chdir(NUTCH_HOME)
  p = os.popen("bin/nutch generate", 'r')
  batch_id = None
  for line in p.readlines():
    print line,
    if line.find("generated batch id") > -1:
      cols = line.split(":")
      if len(cols) == 3:
        batch_id = cols[2].strip()
  p.close()
  os.chdir(cwd)
  return batch_id

def fetch(depth, batch_id):
  print ":::: Fetching (depth %d) with batch_id: %s ::::" % (depth, batch_id)
  cwd = os.getcwd()
  os.chdir(NUTCH_HOME)
  p = os.popen("bin/nutch fetch %s" % (batch_id), 'r')
  for line in p.readlines():
    print line,
  p.close()
  os.chdir(cwd)

def parse(depth, batch_id):
  print ":::: Parsing (depth %d) with batch_id: %s ::::" % (depth, batch_id)
  cwd = os.getcwd()
  os.chdir(NUTCH_HOME)
  p = os.popen("bin/nutch parse %s" % (batch_id), 'r')
  for line in p.readlines():
    print line,
  p.close()
  os.chdir(cwd)

def updatedb(depth):
  print ":::: Update DB (depth %d) ::::" % (depth)
  cwd = os.getcwd()
  os.chdir(NUTCH_HOME)
  p = os.popen("bin/nutch updatedb", 'r')
  for line in p.readlines():
    print line,
  p.close()
  os.chdir(cwd)

def index(depth, batch_id):
  print ":::: Indexing (depth %d) with batch_id: %s ::::" % (depth, batch_id)
  cwd = os.getcwd()
  os.chdir(NUTCH_HOME)
  p = os.popen("bin/nutch solrindex %s %s" % (SOLR_URL, batch_id), 'r')
  for line in p.readlines():
    print line,
  p.close()
  os.chdir(cwd)

def cassandra_stats(casspool, depth):
  count_f = get_rowcount(casspool, "f")
  count_p = get_rowcount(casspool, "p")
  print ":::: In Cassandra (depth %d), count(f) = %d, count(p) = %d ::::" % \
    (depth, count_f, count_p)

def get_rowcount(casspool, cfname):
  cf = pycassa.ColumnFamily(casspool, cfname)
  res = cf.get_range(start="", finish="")
  count = 0
  for k, v in res:
    count = count + 1
  return count

def solr_stats(depth):
  server = "".join([SOLR_URL, "select"])
  params = urllib.urlencode({
    "q" : "*:*", 
    "rows" : "0", 
    "wt" : "json"
  })
  conn = urllib.urlopen(server, params)
  rsp = simplejson.load(conn)
  numfound = rsp["response"]["numFound"]
  print ":::: In Solr (depth %d), %s rows loaded into Solr ::::" % \
    (depth, numfound)

def usage(message=None):
  if message != None:
    print "Error: %s" % (message)
  print "Usage: crawl.py -d depth [-s seed_file]"
  print "       crawl.py -h"
  sys.exit(-1)

def main():
  try:
    (opts, args) = getopt.getopt(sys.argv[1:], 
      "d:s:h", ["depth", "seed", "help"])
  except getopt.GetOptError:
    usage()
  if len(filter(lambda x : x[0] in ("-h", "--help"), opts)) == 1:
    usage()
  crawl_depth = None
  seed = None
  for opt in opts:
    (key, value) = opt
    if key in ("-d", "--depth"):
      crawl_depth = value
    elif key in ("-s", "--seed"):
      seed = value
    else:
      usage()
  # crawl_depth must be specified
  if crawl_depth == None:
    usage("Depth must be specified")
  depth = int(crawl_depth)
  # seed directory if specified must exist
  if seed != None:
    print "seed", seed
    # if seed specified run inject
    if not os.path.isdir(seed):
      usage("Seed directory %s is not a directory" % (seed))
    # TODO: find why this doesn't work?
    #inject(depth, seed)
  # set up pointers to cassandra etc
  casspool = ConnectionPool("webpage", [CASSANDRA_HOST_PORT])
  # for depth iterations run the other methods
  for iter in range(0, int(depth)):
    batch_id = generate(iter)
    fetch(iter, batch_id)
    parse(iter, batch_id)
    updatedb(iter)
    index(iter, batch_id)
    cassandra_stats(casspool, iter)
    solr_stats(iter)

if __name__ == "__main__":
  main()

And finally, I put the project under local git control. I've been using git since my last project, and its pretty addictive :-).

Coding the Plugins

While the Nutch Writing Plugins Wiki page was my main source of information for this work, I also trolled through the Nutch code to find other possible extension points, and here is the list I came up with, so you (and I in about 6 months) don't have to :-).

URLNormalizer = called from nutch inject and generate.
URLFilter - called from nutch generate.
Protocol - called from nutch fetch.
Parser - called from nutch parse.
ScoringFilter - called from nutch updatedb and generate.
IndexingFilter - called from nutch solrindex.

The setup instructions are very detailed, and I followed it almost to the letter to set up my plugin project. Each plugin project has three XML files at its root - the build.xml, ivy.xml and plugin.xml. Here is the Ant build.xml - it delegates all functions to the parent build.xml (supplied by Nutch):

<?xml version="1.0"?>
<1-- Source: src/plugin/mycompany/build.xml -->
<project name="mycompany" default="jar-core">
  <import file="../build-plugin.xml"/>
</project>

And the ivy.xml file:

<?xml version="1.0" encoding="UTF-8"?>
<!-- Source: src/plugin/mycompany/ivy.xml -->
<ivy-module version="1.0">
  
  <info organisation="com.mycompany.nutch" module="${ant.project.name}">
    <license name="Apache 2.0"/>
    <ivyauthor name="Sujit Pal" url="http://sujitpal.blogspot.com"/>
    <description>Nutch Custom Plugins for CNC</description>
  </info>

  <configurations>
    <include file="${nutch.root}/ivy/ivy-configurations.xml"/>
  </configurations>

  <publications>
    <artifact conf="master"/>
  </publications>

  <dependencies>
  </dependencies>

</ivy-module>

And finally, the plugin descriptor plugin.xml file, which describes to the Nutch runtime the capabilities of the plugins in this project, so it can discover and include them at appropriate points in its lifecycle.

<?xml version="1.0" encoding="UTF-8"?>
<!-- Source src/plugin/mycompany/plugin.zml -->
<plugin id="mycompany" 
    name="Custom Company-specific plugins"
    version="1.0.0"
    provider-name="My Company">

  <runtime>
    <library name="mycompany.jar">
      <export name="*"/>
    </library>
  </runtime>

  <requires>
    <import plugin="nutch-extensionpoints"/>
  </requires>

  <extension id="com.mycompany.nutch.indexer.usertags"
      name="User Tag Indexing Filter"
      point="org.apache.nutch.indexer.IndexingFilter">
    <implementation id="mycompany-indexer-usertags"
        class="com.mycompany.nutch.indexer.usertags.UserTagIndexingFilter"/>
  </extension>
  <extension id="com.mycompany.nutch.scoring.usertags"
      name="User Tag Scoring Filter"
      point="org.apache.nutch.scoring.ScoringFilter">
    <implementation id="mycompany-scoring-metadata"
        class="com.mycompany.nutch.scoring.usertags.UserTagScoringFilter"/>
  </extension>

</plugin>

The plugins themselves, as expected, are mostly plain old Java, and have nothing to do with Hadoop or Cassandra - these are abstracted away in the innards of Nutch and GORA. However, the API's specified by the various plugin interfaces are slightly different in the Nutch/GORA case, so I describe them below:

// Source: src/plugin/mycompany/src/java/com/mycompany/nutch/scoring/usertags/UserTagScoringFilter.java
package com.mycompany.nutch.scoring.usertags;

import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.avro.util.Utf8;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.nutch.indexer.NutchDocument;
import org.apache.nutch.scoring.ScoreDatum;
import org.apache.nutch.scoring.ScoringFilter;
import org.apache.nutch.scoring.ScoringFilterException;
import org.apache.nutch.storage.WebPage;
import org.apache.nutch.storage.WebPage.Field;
import org.apache.nutch.util.Bytes;
import org.apache.nutch.util.TableUtil;

public class UserTagScoringFilter implements ScoringFilter {

  private static final Log LOG = LogFactory.getLog(UserTagScoringFilter.class);
  private static final String INHERITABLE_USERTAGS_LIST_PROP = 
    "mycompany.usertags.inheritable";
  private static final Set<WebPage.Field> FIELDS = new HashSet<WebPage.Field>();
  
  static {
    FIELDS.add(WebPage.Field.METADATA);
    FIELDS.add(WebPage.Field.OUTLINKS);
  }
  
  private Configuration conf;
  private Set<String> inheritableUserTags;

  @Override
  public void distributeScoreToOutlinks(String fromUrl, 
      WebPage page, Collection<ScoreDatum> scoreData, 
      int allCount) throws ScoringFilterException {
    // get the usertags from the parent
    LOG.info("Looking for user tags for: " + fromUrl);
    Map<String,String> userTags = new HashMap<String,String>();
    Map<Utf8,ByteBuffer> metadata = page.getMetadata();
    for (Utf8 key : metadata.keySet()) {
      String keyStr = TableUtil.toString(key);
      if (StringUtils.isEmpty(keyStr)) {
        continue;
      }
      if (inheritableUserTags.contains(keyStr)) {
        userTags.put(keyStr, Bytes.toString(
          Bytes.toBytes(metadata.get(key))));
      }
    }
    LOG.info("User tags for " + fromUrl + " = " + userTags);
    // distribute the userTags to all outlink records
    LOG.info("Propagating user tags to outlinks...");
    for (ScoreDatum scoreDatum : scoreData) {
      LOG.info("Processing outlink:" + scoreDatum.getUrl());
      for (String tagName : userTags.keySet()) {
        String tagValue = userTags.get(tagName);
        if (StringUtils.isNotEmpty(tagValue)) {
          LOG.info("Setting outlink.meta[" + tagName + 
              "]=" + tagValue + " for outlink:" + 
              scoreDatum.getUrl());
          scoreDatum.setMeta(tagName, Bytes.toBytes(tagValue));
        }
      }
    }
  }

  @Override
  public void updateScore(String url, WebPage page,
      List<ScoreDatum> inlinkedScoreData) 
      throws ScoringFilterException {
    // these are the outlinks, updateScore is called once
    // per outlink to which we propagated metadata to in
    // distributeScoreToOutlinks, so this is our chance
    // to persist it there...
    // extract all user tags from inlinks
    Map<Utf8,ByteBuffer> metadata = new HashMap<Utf8,ByteBuffer>();
    for (ScoreDatum scoreDatum : inlinkedScoreData) {
      LOG.info("Looking at inlink: " + scoreDatum.getUrl());
      for (String inheritableUserTag : inheritableUserTags) {
        String value = Bytes.toString(
          scoreDatum.getMeta(inheritableUserTag));
        if (StringUtils.isNotEmpty(value)) {
          metadata.put(new Utf8(inheritableUserTag), 
            ByteBuffer.wrap(Bytes.toBytes(value)));
        }
      }
    }
    if (metadata.size() > 0) {
      for (Utf8 key : metadata.keySet()) {
        LOG.info("Saving metadata[" + TableUtil.toString(key) + 
          "] to outlink: " + url);
        page.putToMetadata(key, metadata.get(key));
      }
    }
    return;
  }

  @Override
  public float generatorSortValue(String url, WebPage page, 
      float initSort) throws ScoringFilterException {
    return initSort;
  }

  @Override
  public float indexerScore(String url, NutchDocument doc, 
      WebPage page, float initScore) 
      throws ScoringFilterException {
    return 0;
  }

  @Override
  public void initialScore(String url, WebPage page)
      throws ScoringFilterException {
    return;
  }

  @Override
  public void injectedScore(String url, WebPage page)
      throws ScoringFilterException {
    return;
  }

  @Override
  public Configuration getConf() {
    return conf;
  }

  @Override
  public void setConf(Configuration conf) {
    this.conf = conf;
    inheritableUserTags = new HashSet<String>(
      Arrays.asList(StringUtils.split(
      conf.get(INHERITABLE_USERTAGS_LIST_PROP, ""), ",")));
    LOG.info("Property " + INHERITABLE_USERTAGS_LIST_PROP + 
      " set to " + inheritableUserTags);
  }

  @Override
  public Collection<Field> getFields() {
    return FIELDS;
  }
}

The two main methods are the distributeScoreToOutlinks and the updateScore methods (although we are really distributing and updating metadata rather than scores here). The first one is called by the DbUpdaterMapper and is used to "scatter" the parent metadata to its outlinks, and the second is called by the DbUpdaterReducer and is used to "gather" back the metadata into the child page from eac of its parent inlinks. Its a bit counter-intuitive to have this happen in two steps, but I think the reason is that in the mapper you don't have a handle to the child WebPages, and the ScoreDatum is just for passing to the reducer and is not written to Cassandra. Additionally, the get/setConf() methods are to extract the property from nutch-site.xml. The getFields() as far as I could make out, work a bit like Lucene's FieldSelectors.

The second plugin is the indexing filter. Its main method is filter(), which extracts the metadata from the webpage["sc"]["mtdt"] and places it into the NutchDocument for publishing to the index. Here is the code, its relatively simple.

// Source: src/plugin/mycompany/src/java/com/mycompany/nutch/indexer/usertags/UserTagIndexingFilter.java
package com.mycompany.nutch.indexer.usertags;

import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.avro.util.Utf8;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.nutch.indexer.IndexingException;
import org.apache.nutch.indexer.IndexingFilter;
import org.apache.nutch.indexer.NutchDocument;
import org.apache.nutch.storage.WebPage;
import org.apache.nutch.storage.WebPage.Field;
import org.apache.nutch.util.Bytes;
import org.apache.nutch.util.TableUtil;

public class UserTagIndexingFilter implements IndexingFilter {

  private static final Log LOG = LogFactory.getLog(UserTagIndexingFilter.class);
  private static final String INHERITABLE_USERTAGS_LIST_PROP = 
    "mycompany.usertags.inheritable";
  
  private static final Set<WebPage.Field> FIELDS = new HashSet<WebPage.Field>();
  
  static {
    FIELDS.add(WebPage.Field.METADATA);
  }
  
  private Configuration conf;
  private Set<String> inheritableUserTags;
  
  @Override
  public NutchDocument filter(NutchDocument doc, String url, 
      WebPage page) throws IndexingException {
    LOG.info("Adding inheritable user tags for page:" + url);
    Map<Utf8,ByteBuffer> metadata = page.getMetadata();
    for (Utf8 key : metadata.keySet()) {
      String keyStr = TableUtil.toString(key);
      if (StringUtils.isEmpty(keyStr)) {
        continue;
      }
      if (inheritableUserTags.contains(keyStr)) {
        String value = Bytes.toString(
          Bytes.toBytes(metadata.get(key)));
        doc.add(keyStr, value);
      }
    }
    return doc;
  }

  @Override
  public Configuration getConf() {
    return conf;
  }

  @Override
  public void setConf(Configuration conf) {
    this.conf = conf;
    inheritableUserTags = new HashSet<String>(
      Arrays.asList(StringUtils.split(
      conf.get(INHERITABLE_USERTAGS_LIST_PROP, ""), ",")));
    LOG.info("Property " + INHERITABLE_USERTAGS_LIST_PROP + 
      " set to " + inheritableUserTags);
  }

  @Override
  public Collection<Field> getFields() {
    return FIELDS;
  }
}

In addition, we need to add the u_idx field into the conf/solrindex-mapping.xml file so Nutch knows to index the field, like so:

<?xml version="1.0" encoding="UTF-8"?>
<!-- Source: conf/solrindex-mapping.xml -->
<mapping>
  <fields>
    ...
    <field dest="u_idx" source="u_idx"/>
  </fields>
  <uniqueKey>id</uniqueKey>
</mapping>

and we also need to add this into the schema.xml file that will go into the Solr instance.

1	<field name="u_idx" type="string" stored="true" indexed="true"/>

And thats pretty much it, running the crawl to a depth of 3 against the test site produces 7 pages in the Solr index, as expected, and all 7 of these pages have u_idx set to "test".

Exploring Nutch-GORA with Cassandra

2012-01-07T18:17:00.000-08:00

Background

I started looking at this early last year, with a view to building an incremental indexing pipeline to build our Lucene indexes. For most of last year, a group of us have been building an integrated content ingestion and search service around our proprietary concept search algorithms using Cassandra and Solr. During the initial design phase, I proposed a design for the indexing pipeline that was kind of where I wanted to go with Nutch/GORA. However, my main focus for the project was Solr and search, and for various reasons, the backend did not turn out quite the way I had hoped.

While I worked on Solr quite heavily and learned a lot over the past year, I did not get time to work on (and learn how to use) Cassandra. Its biting me a bit at the moment, now that the project is winding down and we are uncovering integration issues - our index is now quite large, and many times, the issues can be fixed long-term by fixing the parsing code, and (additionally) short-term by updating the data in Cassandra and republishing to Solr. Being able to do the latter would have significantly decreased turnaround time in quite a few cases.

So I decided to resurrect this project (using Nutch/GORA with Cassandra to build a pipeline that would work with content from different sources - web crawls, feeds, content management systems and semi-structured content (XML/JSON) from external providers, and incrementally index them into a single large Solr index). Mainly because I believe this would be a nice generic solution, but a nice side effect would be gaining familiarity with Cassandra, as well as Nutch, GORA and Hadoop. So, hopefully (unless I get sidetracked again by something shinier :-)), the next few posts would be about this project.

In this post, I follow a path similar to my previous experiment with Nutch/GORA (previously Nutch 2.0 trunk, now in its own branch) and HBase, basically running the various nutch sub-commands to crawl a small external site and finally building the index using Nutch's solrindex sub-command. At each stage, I run some Python/Pycassa scripts to see what went into the Cassandra database, mainly to understand what each subcommand does, so I can modify/extend the behavior for my purposes if required.

Setup

For setup, I followed this excellent Tech Vineyard post almost to the letter. Broadly here are the steps:

Download and install Cassandra - I downloaded Cassandra version 1.0.6 from here. Installation is simply untarring into some location on disk.
Download Nutch from the NutchGora branch and run the default ant target, this creates the runtime/local subdirectory which can be used to run Nutch.
Download GORA from the GORA trunk. To compile this, I couldn't get the ant/ivy setup to work, but I could compile the project and build the GORA JAR using mvn.
Follow the instructions in this TechVineyard post to set up Nutch with Cassandra. Of particular importance is setting the storage engine in nutch-site.xml and making sure that the gora-cassandra-mapping.xml file is in your runtime/local/conf directory.
Set the http.agent.name (your crawler name) and the http.robot.agents in nutch-default.xml in runtime/local/conf, otherwise nutch will complain when you try to run it.
Download Hector, a Cassandra Java client API which seems to be used by GORA. We need it for some of the library dependecies. I could have probably got these libraries from GORA if I could get ivy to work, but I couldn't, so I had to get these libraries from Hector.
In the runtime/local/lib directory, replace cassandra-thrift-0.8.jar with cassandra-thrift-1.0.1.jar from the Cassandra distribution. From the Hector distribution, copy over the hector-core-1.0-1.jar and guava-r09.jar. Also copy over the gora-cassandra-0.2-SNAPSHOT.jar (and perhaps the other gora JARs as well, but so far I haven't had to).
Download Solr from the Solr download page. I got Solr 3.5.0. Copy over the runtime/local/conf/schema.xml to Solr's examples/solr/conf directory. This replaces Solr's default schema with Nutch's.
Download Pycassa from the Pycassa download page - I needed this to write little scripts to peek inside Cassandra as I executed the various nutch subcommands. Installation instructions can be found here.
Start Cassandra with $CASSANDRA_HOME/bin/cassandra -f. This sets up a Cassandra server on port 9160. Cassandra can be terminated with Ctrl-C.
Start Solr with java -jar start.jar from the solr/example directory. This starts up a Jetty server running Solr on port 8983.

At this point we are ready to run the nutch subcommands and observe the effects of running these commands in the Cassandra database.

Nutch Inject

The nutch inject subcommand allows you to add a list of seed URLs for your crawl. For my test, I used a small site I was familiar with from earlier crawler testing. Nutch expects the seed URLs to be supplied as a directory of flat files. Each line in a flat file consists of the seed URL, optionally followed by name-value metadata elements. Here is my seed list file.

1 2	# Source: web_seeds/parathyroid.com http://www.parathyroid.com u_idx=web

I want all pages crawled from this seed to have an extra metadata attribute u_idx set to "web". This is so I can differentiate between crawled content and other sources in my Solr index.

To inject this URL we run the following command. This command takes the URLs from the seed list(s) and puts them on Nutch's list of URLs to fetch.

sujit@cyclone:local$ bin/nutch inject ../../../web_seeds
InjectorJob: starting
InjectorJob: urlDir: ../../../web_seeds
InjectorJob: finished

At this point Cassandra has created a keyspace "webpage", and its "f" column family (fetched content) contains a single record as shown below (this is the output of the display_webpage.py script, which is provided further down). Notice that our user supplied metadata "u_idx" is available in the webpage["sc"]["mtdt"]["u_idx"] column.

webpage: {
.. key: "com.parathyroid.www:http/" ,
.. f: {
.... fi : "2592000" ,
.... s : "1.0" ,
.... ts : "1325623018630" ,
.. },
.. p: {
.. },
.. sc: {
.... mk : {
...... _injmrk_ : "y" ,
.... }
.... mtdt : {
...... _csh_ : "?" ,
...... u_idx : "web" ,
.... }
.. }
}

Nutch Generate

The nutch generate command will take the list of outlinks generated from a previous cycle and promote them to the fetch list and return a batch ID for this cycle. You will need this batch ID for subsequent calls in this cycle. Here's the command:

sujit@cyclone:local$ bin/nutch generate
GeneratorJob: Selecting best-scoring urls due for fetch.
GeneratorJob: starting
GeneratorJob: filtering: true
GeneratorJob: done
GeneratorJob: generated batch id: 1325623123-1500197250

Once more we look at what's in Cassandra at the end of this run. Notice that not much has changed, except it created a webpage["sc"]["mk"]["_gnmrk_"] record with the batch ID it spat out.

webpage: {
.. key: "com.parathyroid.www:http/" ,
.. f: {
.... fi : "2592000" ,
.... s : "1.0" ,
.... ts : "1325623018630" ,
.. },
.. p: {
.. },
.. sc: {
.... mk : {
...... _gnmrk_ : "1325623123-1500197250" ,
...... _injmrk_ : "y" ,
.... }
.... mtdt : {
...... _csh_ : "?" ,
...... u_idx : "web" ,
.... }
.. }
}

Nutch Fetch

Next we execute a fetch. The Nutch Fetch command will crawl the pages listed in the "f" column family and write out the contents into new columns in "f". We need to pass in the batch ID from the previous step. The command looks like this:

sujit@cyclone:local$ bin/nutch fetch 1325623123-1500197250
FetcherJob: starting
FetcherJob : timelimit set for : -1
FetcherJob: threads: 10
FetcherJob: parsing: false
FetcherJob: resuming: false
FetcherJob: batchId: 1325623123-1500197250
Using queue mode : byHost
Fetcher: threads: 10
QueueFeeder finished: total 1 records. Hit by time limit :0
fetching http://www.parathyroid.com/
-finishing thread FetcherThread1, activeThreads=1
-finishing thread FetcherThread2, activeThreads=1
-finishing thread FetcherThread3, activeThreads=1
-finishing thread FetcherThread4, activeThreads=1
-finishing thread FetcherThread5, activeThreads=1
-finishing thread FetcherThread6, activeThreads=1
-finishing thread FetcherThread7, activeThreads=1
-finishing thread FetcherThread8, activeThreads=1
-finishing thread FetcherThread9, activeThreads=1
-finishing thread FetcherThread0, activeThreads=0
-activeThreads=0, spinWaiting=0, fetchQueues= 0, fetchQueues.totalSize=0
-activeThreads=0
FetcherJob: done

As you can see, the fetch command wrote out new columns in the "f" and "sc" column families. These new columns are additional data and page metadata found during the fetch. But since we started with only one injected URL in "f", there is still only one record in "f".

webpage: {
.. key: "com.parathyroid.www:http/" ,
.. f: {
.... bas : "http://www.parathyroid.com/" ,
.... cnt : "<html>content omitted here</html>",
.... fi : "2592000" ,
.... s : "1.0" ,
.... st : "2" ,
.... ts : "1325623263386" ,
.... typ : "text/html" ,
.. },
.. p: {
.. },
.. sc: {
.... h : {
...... Accept-Ranges : "bytes" ,
...... Connection : "close" ,
...... Content-Length : "64557" ,
...... Content-Location : "http://www.parathyroid.com/index.htm" ,
...... Content-Type : "text/html" ,
...... Date : "Tue, 03 Jan 2012 20:41:02 GMT" ,
...... ETag : None ,
...... Last-Modified : "Fri, 30 Dec 2011 20:20:45 GMT" ,
...... MicrosoftOfficeWebServer : "5.0_Pub" ,
...... Server : "Microsoft-IIS/6.0" ,
...... X-Powered-By : "ASP.NET" ,
.... }
.... mk : {
...... _ftcmrk_ : "1325623123-1500197250" ,
...... _gnmrk_ : "1325623123-1500197250" ,
...... _injmrk_ : "y" ,
.... }
.... mtdt : {
...... _csh_ : "?" ,
...... u_idx : "web" ,
.... }
.... prs : {
...... code : "1" ,
...... lastModified : "0" ,
.... }
.. }
}

Nutch Parse

The nutch parse subcommand will loop through all the pages in the "f" column family (one in our case), analyze the page content to find outgoing links, and write them out in the "p" column family. Here is the command:

sujit@cyclone:local$ bin/nutch parse 1325623124-1500197250
ParserJob: starting
ParserJob: resuming: false
ParserJob: forced reparse: false
ParserJob: batchId: 1325623123-1500197250
Parsing http://www.parathyroid.com/
ParserJob: success

And the data inside Cassandra now looks like this:

webpage: {
.. key: "com.parathyroid.www:http/" ,
.. f: {
.... bas : "http://www.parathyroid.com/" ,
.... cnt : "<html>HTML Content ommitted</html>",
.... fi : "2592000" ,
.... s : "1.0" ,
.... st : "2" ,
.... ts : "1325623263386" ,
.... typ : "text/html" ,
.. },
.. p: {
.... c : "a very long plaintext string representing parsed content",
.... sig : "gm" ,
.... t : "Title of document extracted from HTML/HEAD/TITLE tag",
.. },
.. sc: {
.... h : {
...... Accept-Ranges : "bytes" ,
...... Connection : "close" ,
...... Content-Length : "64557" ,
...... Content-Location : "http://www.parathyroid.com/index.htm" ,
...... Content-Type : "text/html" ,
...... Date : "Tue, 03 Jan 2012 20:41:02 GMT" ,
...... ETag : None ,
...... Last-Modified : "Fri, 30 Dec 2011 20:20:45 GMT" ,
...... MicrosoftOfficeWebServer : "5.0_Pub" ,
...... Server : "Microsoft-IIS/6.0" ,
...... X-Powered-By : "ASP.NET" ,
.... }
.... mk : {
...... __prsmrk__ : "1325623123-1500197250" ,
...... _ftcmrk_ : "1325623123-1500197250" ,
...... _gnmrk_ : "1325623123-1500197250" ,
...... _injmrk_ : "y" ,
.... }
.... mtdt : {
...... _csh_ : "?" ,
...... u_idx : "web" ,
.... }
.... ol : {
...... http://www.parathyroid.com/ : "" ,
...... http://www.parathyroid.com/10_parathyroid_rules.htm : "10ParathyroidRules" ,
...... http://www.parathyroid.com/Dr.Norman.htm : "Dr.Norman" ,
...... http://www.parathyroid.com/FAQ.htm : "FrequentQuestions" ,
...... http://www.parathyroid.com/FHH.htm : "FHH" ,
...... http://www.parathyroid.com/MEN-Syndrome.htm : "MEN Syndromes" ,
...... http://www.parathyroid.com/MIRP-Surgery.htm : "MIRPMiniSurgery" ,
...... http://www.parathyroid.com/MIRP-publications.htm : "Publications" ,
...... http://www.parathyroid.com/Parathyroid-Surgeon-Map.htm : "PatientMap" ,
...... http://www.parathyroid.com/Parathyroid-Surgeon.htm : "BecomeAPatient" ,
...... http://www.parathyroid.com/Re-Operation.htm : "Re-Operate" ,
...... http://www.parathyroid.com/Sensipar-high-calcium.htm : "Sensipar" ,
...... http://www.parathyroid.com/about-Parathyroid.htm : "Norman Parathyroid Center" ,
...... http://www.parathyroid.com/age.htm : "WhoGetsIt?" ,
...... http://www.parathyroid.com/causes.htm : "WhatCausesIt?" ,
...... http://www.parathyroid.com/contents.htm : "TableofContents" ,
...... http://www.parathyroid.com/diagnosis.htm : "Diagnosis" ,
...... http://www.parathyroid.com/disclaimer.htm : "Disclaimer" ,
...... http://www.parathyroid.com/endocrinology.htm : "WhatExpertsSay" ,
...... http://www.parathyroid.com/finding-parathyroid.htm : "FindingtheTumor" ,
...... http://www.parathyroid.com/high-calcium.htm : "HighBloodCalcium" ,
...... http://www.parathyroid.com/hyperparathyroidism-diagnosis.htm : "Diagnosis-ADVANCED" ,
...... http://www.parathyroid.com/hyperparathyroidism-videos.htm : "TeachingVideos" ,
...... http://www.parathyroid.com/hypoparathyroidism.htm : "Hyp0parathyroid" ,
...... http://www.parathyroid.com/index.htm : "" ,
...... http://www.parathyroid.com/low-calcium.htm : "Surgeon-Induced HypOparathyroidism." ,
...... http://www.parathyroid.com/low-vitamin-d.htm : "LowVitaminD" ,
...... http://www.parathyroid.com/mini-surgery.htm : "Mini-Surgery" ,
...... http://www.parathyroid.com/osteoporosis.htm : "Osteoporosis" ,
...... http://www.parathyroid.com/osteoporosis2.htm : "Life insurance companies" ,
...... http://www.parathyroid.com/parathyroid-adenoma.htm : "DoIHaveJustOne?" ,
...... http://www.parathyroid.com/parathyroid-anatomy.htm : "Parathyroid Anatomy" ,
...... http://www.parathyroid.com/parathyroid-cancer.htm : "ParathyroidCancer" ,
...... http://www.parathyroid.com/parathyroid-disease.htm : "Hyperparathyroidism" ,
...... http://www.parathyroid.com/parathyroid-function.htm : "NormalFunction" ,
...... http://www.parathyroid.com/parathyroid-pictures.htm : "ParathyroidPictures" ,
...... http://www.parathyroid.com/parathyroid-surgery.htm : "SurgeryVideo" ,
...... http://www.parathyroid.com/parathyroid-symptoms-cartoon.htm : "SymptomCartoon" ,
...... http://www.parathyroid.com/parathyroid-symptoms.htm : "Symptoms" ,
...... http://www.parathyroid.com/parathyroid.htm : "ParathyroidIntro" ,
...... http://www.parathyroid.com/paratiroide : "" ,
...... http://www.parathyroid.com/paratiroide/index.html : "Espanol/Spanish" ,
...... http://www.parathyroid.com/pregnancy.htm : "Parathyroid disease during Pregnancy." ,
...... http://www.parathyroid.com/sestamibi.htm : "SestamibiScan" ,
...... http://www.parathyroid.com/surgery_cure_rates.htm : "SurgeryCureRates" ,
...... http://www.parathyroid.com/testimonials.htm : "WhatPatientsSay" ,
...... http://www.parathyroid.com/treatment-surgery.htm : "Treatment/Surgery" ,
...... http://www.parathyroid.com/who's_eligible.htm : "Who is Eligible to Have a MIRP Mini-Parathyroid Operation?" ,
.... }
.... pas : {
...... majorCode : "1" ,
...... minorCode : "0" ,
.... }
.... prs : {
...... code : "1" ,
...... lastModified : "0" ,
.... }
.. }
}

The parse command parses the fetched page content in webpage["f"]["cnt"] and places the parsed output in webpage["p"]["cnt"]. In addition, it places the inline HTML links in the page into the webpage["sc"]["ol"] column.

At this point there is still only 1 record (the seed URL) in Cassandra.

Nutch UpdateDB

The nutch updatedb subcommand takes the webpage["sc"]["ol"] values from the previous stage and places it into the webpage["f"] column family, so they can be fetched in the next crawl cycle.

1
2
3

sujit@cyclone:local$ bin/nutch updatedb
DbUpdaterJob: starting
DbUpdaterJob: done

At this point, I switched to counting the number of columns in the "f" and "p" column families, since the JSON style output I was doing would be too large to consume. At the end of the updatedb, I have the following URLs in the "f" and "p" column families.

sujit@cyclone:scripts$ ./count_pages.py f
com.parathyroid.www:http/endocrinology.htm
com.parathyroid.www:http/causes.htm
com.parathyroid.www:http/MEN-Syndrome.htm
com.parathyroid.www:http/Parathyroid-Surgeon.htm
com.parathyroid.www:http/Sensipar-high-calcium.htm
com.parathyroid.www:http/who's_eligible.htm
com.parathyroid.www:http/parathyroid-cancer.htm
com.parathyroid.www:http/mini-surgery.htm
com.parathyroid.www:http/surgery_cure_rates.htm
com.parathyroid.www:http/
com.parathyroid.www:http/age.htm
com.parathyroid.www:http/index.htm
com.parathyroid.www:http/10_parathyroid_rules.htm
com.parathyroid.www:http/parathyroid-function.htm
com.parathyroid.www:http/contents.htm
com.parathyroid.www:http/FAQ.htm
com.parathyroid.www:http/paratiroide
com.parathyroid.www:http/MIRP-publications.htm
com.parathyroid.www:http/Dr.Norman.htm
com.parathyroid.www:http/MIRP-Surgery.htm
com.parathyroid.www:http/parathyroid-adenoma.htm
com.parathyroid.www:http/parathyroid-disease.htm
com.parathyroid.www:http/FHH.htm
com.parathyroid.www:http/treatment-surgery.htm
com.parathyroid.www:http/parathyroid-surgery.htm
com.parathyroid.www:http/hyperparathyroidism-diagnosis.htm
com.parathyroid.www:http/Parathyroid-Surgeon-Map.htm
com.parathyroid.www:http/hypoparathyroidism.htm
com.parathyroid.www:http/parathyroid-symptoms.htm
com.parathyroid.www:http/parathyroid-symptoms-cartoon.htm
com.parathyroid.www:http/parathyroid-pictures.htm
com.parathyroid.www:http/osteoporosis.htm
com.parathyroid.www:http/parathyroid.htm
com.parathyroid.www:http/about-Parathyroid.htm
com.parathyroid.www:http/Re-Operation.htm
com.parathyroid.www:http/diagnosis.htm
com.parathyroid.www:http/pregnancy.htm
com.parathyroid.www:http/high-calcium.htm
com.parathyroid.www:http/parathyroid-anatomy.htm
com.parathyroid.www:http/low-vitamin-d.htm
com.parathyroid.www:http/testimonials.htm
com.parathyroid.www:http/low-calcium.htm
com.parathyroid.www:http/finding-parathyroid.htm
com.parathyroid.www:http/hyperparathyroidism-videos.htm
com.parathyroid.www:http/paratiroide/index.html
com.parathyroid.www:http/sestamibi.htm
com.parathyroid.www:http/osteoporosis2.htm
com.parathyroid.www:http/disclaimer.htm
#-records in f: 48

sujit@cyclone:scripts$ ./count_pages.py p
com.parathyroid.www:http/
#-records in p: 1

Iterating

We then iterate through the generate, fetch, parse, and updatedb subcommand two more times to crawl the site to a depth of 3. Shown below is the nutch console logs for the second iteration.

sujit@cyclone:local$ bin/nutch generate
GeneratorJob: Selecting best-scoring urls due for fetch.
GeneratorJob: starting
GeneratorJob: filtering: true
GeneratorJob: done
GeneratorJob: generated batch id: 1325709400-776802111
sujit@cyclone:local$ bin/nutch fetch 1325709400-776802111
FetcherJob: starting
FetcherJob : timelimit set for : -1
FetcherJob: threads: 10
FetcherJob: parsing: false
FetcherJob: resuming: false
FetcherJob: batchId: 1325709400-776802111
Using queue mode : byHost
Fetcher: threads: 10
fetching http://www.parathyroid.com/parathyroid.htm
QueueFeeder finished: total 47 records. Hit by time limit :0
-activeThreads=10, spinWaiting=9, fetchQueues= 1, fetchQueues.totalSize=46
fetching http://www.parathyroid.com/Parathyroid-Surgeon.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=45
fetching http://www.parathyroid.com/paratiroide/index.html
fetching http://www.parathyroid.com/diagnosis.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=43
fetching http://www.parathyroid.com/parathyroid-adenoma.htm
fetching http://www.parathyroid.com/age.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=41
fetching http://www.parathyroid.com/FHH.htm
-activeThreads=10, spinWaiting=9, fetchQueues= 1, fetchQueues.totalSize=40
fetching http://www.parathyroid.com/treatment-surgery.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=39
fetching http://www.parathyroid.com/who's_eligible.htm
fetching http://www.parathyroid.com/parathyroid-disease.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=37
fetching http://www.parathyroid.com/FAQ.htm
fetching http://www.parathyroid.com/finding-parathyroid.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=35
fetching http://www.parathyroid.com/hyperparathyroidism-diagnosis.htm
-activeThreads=10, spinWaiting=9, fetchQueues= 1, fetchQueues.totalSize=34
fetching http://www.parathyroid.com/index.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=33
fetching http://www.parathyroid.com/parathyroid-pictures.htm
fetching http://www.parathyroid.com/Parathyroid-Surgeon-Map.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=31
fetching http://www.parathyroid.com/mini-surgery.htm
fetching http://www.parathyroid.com/about-Parathyroid.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=29
fetching http://www.parathyroid.com/disclaimer.htm
-activeThreads=10, spinWaiting=9, fetchQueues= 1, fetchQueues.totalSize=28
fetching http://www.parathyroid.com/parathyroid-function.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=27
fetching http://www.parathyroid.com/paratiroide
fetching http://www.parathyroid.com/low-vitamin-d.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=25
fetching http://www.parathyroid.com/parathyroid-symptoms-cartoon.htm
fetching http://www.parathyroid.com/sestamibi.htm
-activeThreads=10, spinWaiting=9, fetchQueues= 1, fetchQueues.totalSize=23
fetching http://www.parathyroid.com/osteoporosis.htm
-activeThreads=10, spinWaiting=9, fetchQueues= 1, fetchQueues.totalSize=22
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=22
fetching http://www.parathyroid.com/surgery_cure_rates.htm
fetching http://www.parathyroid.com/low-calcium.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=20
fetching http://www.parathyroid.com/Sensipar-high-calcium.htm
fetching http://www.parathyroid.com/Dr.Norman.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=18
fetching http://www.parathyroid.com/parathyroid-anatomy.htm
fetching http://www.parathyroid.com/parathyroid-surgery.htm
-activeThreads=10, spinWaiting=9, fetchQueues= 1, fetchQueues.totalSize=16
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=16
fetching http://www.parathyroid.com/hypoparathyroidism.htm
fetching http://www.parathyroid.com/endocrinology.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=14
fetching http://www.parathyroid.com/parathyroid-cancer.htm
fetching http://www.parathyroid.com/testimonials.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=12
fetching http://www.parathyroid.com/hyperparathyroidism-videos.htm
fetching http://www.parathyroid.com/high-calcium.htm
-activeThreads=10, spinWaiting=9, fetchQueues= 1, fetchQueues.totalSize=10
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=10
fetching http://www.parathyroid.com/osteoporosis2.htm
fetching http://www.parathyroid.com/MEN-Syndrome.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=8
fetching http://www.parathyroid.com/causes.htm
fetching http://www.parathyroid.com/MIRP-Surgery.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=6
fetching http://www.parathyroid.com/Re-Operation.htm
fetching http://www.parathyroid.com/pregnancy.htm
-activeThreads=10, spinWaiting=9, fetchQueues= 1, fetchQueues.totalSize=4
* queue: http://www.parathyroid.com
  maxThreads    = 1
  inProgress    = 1
  crawlDelay    = 5000
  minCrawlDelay = 0
  nextFetchTime = 1325709723044
  now           = 1325709724390
  0. http://www.parathyroid.com/parathyroid-symptoms.htm
  1. http://www.parathyroid.com/10_parathyroid_rules.htm
  2. http://www.parathyroid.com/MIRP-publications.htm
  3. http://www.parathyroid.com/contents.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=4
* queue: http://www.parathyroid.com
  maxThreads    = 1
  inProgress    = 0
  crawlDelay    = 5000
  minCrawlDelay = 0
  nextFetchTime = 1325709738263
  now           = 1325709734391
  0. http://www.parathyroid.com/parathyroid-symptoms.htm
  1. http://www.parathyroid.com/10_parathyroid_rules.htm
  2. http://www.parathyroid.com/MIRP-publications.htm
  3. http://www.parathyroid.com/contents.htm
fetching http://www.parathyroid.com/parathyroid-symptoms.htm
fetching http://www.parathyroid.com/10_parathyroid_rules.htm
-activeThreads=10, spinWaiting=10, fetchQueues= 1, fetchQueues.totalSize=2
* queue: http://www.parathyroid.com
  maxThreads    = 1
  inProgress    = 0
  crawlDelay    = 5000
  minCrawlDelay = 0
  nextFetchTime = 1325709748664
  now           = 1325709744393
  0. http://www.parathyroid.com/MIRP-publications.htm
  1. http://www.parathyroid.com/contents.htm
fetching http://www.parathyroid.com/MIRP-publications.htm
fetching http://www.parathyroid.com/contents.htm
-finishing thread FetcherThread5, activeThreads=9
-finishing thread FetcherThread3, activeThreads=8
-finishing thread FetcherThread2, activeThreads=7
-finishing thread FetcherThread1, activeThreads=6
-finishing thread FetcherThread8, activeThreads=5
-finishing thread FetcherThread7, activeThreads=2
-finishing thread FetcherThread9, activeThreads=3
-finishing thread FetcherThread6, activeThreads=4
-finishing thread FetcherThread4, activeThreads=1
-finishing thread FetcherThread0, activeThreads=0
-activeThreads=0, spinWaiting=0, fetchQueues= 0, fetchQueues.totalSize=0
-activeThreads=0
FetcherJob: done
ujit@cyclone:local$ bin/nutch parse 1325709400-776802111
ParserJob: starting
ParserJob: resuming: false
ParserJob: forced reparse: false
ParserJob: batchId: 1325709400-776802111
Parsing http://www.parathyroid.com/endocrinology.htm
Parsing http://www.parathyroid.com/causes.htm
Parsing http://www.parathyroid.com/MEN-Syndrome.htm
Parsing http://www.parathyroid.com/Parathyroid-Surgeon.htm
Parsing http://www.parathyroid.com/Sensipar-high-calcium.htm
Parsing http://www.parathyroid.com/who's_eligible.htm
Parsing http://www.parathyroid.com/parathyroid-cancer.htm
Parsing http://www.parathyroid.com/mini-surgery.htm
Parsing http://www.parathyroid.com/surgery_cure_rates.htm
Skipping http://www.parathyroid.com/; different batch id
Parsing http://www.parathyroid.com/age.htm
Parsing http://www.parathyroid.com/index.htm
Parsing http://www.parathyroid.com/10_parathyroid_rules.htm
Parsing http://www.parathyroid.com/parathyroid-function.htm
Parsing http://www.parathyroid.com/contents.htm
Parsing http://www.parathyroid.com/FAQ.htm
Parsing http://www.parathyroid.com/paratiroide
Parsing http://www.parathyroid.com/MIRP-publications.htm
Parsing http://www.parathyroid.com/Dr.Norman.htm
Parsing http://www.parathyroid.com/MIRP-Surgery.htm
Parsing http://www.parathyroid.com/parathyroid-adenoma.htm
Parsing http://www.parathyroid.com/parathyroid-disease.htm
Parsing http://www.parathyroid.com/FHH.htm
Parsing http://www.parathyroid.com/treatment-surgery.htm
Parsing http://www.parathyroid.com/parathyroid-surgery.htm
Parsing http://www.parathyroid.com/hyperparathyroidism-diagnosis.htm
Parsing http://www.parathyroid.com/Parathyroid-Surgeon-Map.htm
Parsing http://www.parathyroid.com/hypoparathyroidism.htm
Parsing http://www.parathyroid.com/parathyroid-symptoms.htm
Skipping http://www.parathyroid.com/paratiroide/; different batch id
Parsing http://www.parathyroid.com/parathyroid-symptoms-cartoon.htm
Parsing http://www.parathyroid.com/parathyroid-pictures.htm
Parsing http://www.parathyroid.com/osteoporosis.htm
Parsing http://www.parathyroid.com/parathyroid.htm
Parsing http://www.parathyroid.com/about-Parathyroid.htm
Parsing http://www.parathyroid.com/Re-Operation.htm
Parsing http://www.parathyroid.com/diagnosis.htm
Parsing http://www.parathyroid.com/pregnancy.htm
Parsing http://www.parathyroid.com/high-calcium.htm
Parsing http://www.parathyroid.com/parathyroid-anatomy.htm
Parsing http://www.parathyroid.com/low-vitamin-d.htm
Parsing http://www.parathyroid.com/testimonials.htm
Parsing http://www.parathyroid.com/low-calcium.htm
Parsing http://www.parathyroid.com/finding-parathyroid.htm
Parsing http://www.parathyroid.com/hyperparathyroidism-videos.htm
Parsing http://www.parathyroid.com/paratiroide/index.html
Parsing http://www.parathyroid.com/sestamibi.htm
Parsing http://www.parathyroid.com/osteoporosis2.htm
Parsing http://www.parathyroid.com/disclaimer.htm
ParserJob: success
sujit@cyclone:local$ bin/nutch updatedb
DbUpdaterJob: starting
DbUpdaterJob: done

As you can see, nutch is now fetching and parsing all the URLs that have been put by the updatedb subcommand in its fetch list. I recorded the number of records in The "f" and "p" column families after each iteration, as shown below:

after second iteration: count(f)=149, count(p) = 47
after third iteration, count(f)=627, count(p) = 108

Nutch SolrIndex

It is possible (and probably recommended, for incremental indexing) to publish the parsed records to Solr by batch ID after each iteration, but I decided to do this after my entire crawl was done. I already had a Solr server listening on port 8983 with the nutch schema.xml file. Here's the nutch subcommand I used.

1
2
3

sujit@cyclone:local$ bin/nutch solrindex http://127.0.0.1:8983/solr/ -all
SolrIndexerJob: starting
SolrIndexerJob: done.

Navigating to http://localhost:8983/solr/admin and issuing a *:* (match all records) gives me back 108 records, which matches the count of records in the "p" column family after the third crawl loop iteration.

Pycassa Scripts

And here are the Python scripts for display_webpage.py and count_pages.py I promised earlier. The display_webpage.py reads all the records in the webpage keyspace and dumps them out in easy to understand JSON-like format to the console.

#!/usr/bin/python

import pycassa
from pycassa.pool import ConnectionPool
from pycassa.util import OrderedDict
import sys

def print_map(level, dict):
  for key in dict.keys():
    value = dict[key]
    if type(value) == type(OrderedDict()):
      print indent(level), key, ": {"
      print_map(level+1, value)
      print indent(level), "}"
    else:
      print indent(level), key, ":", quote(value), ","
    
def quote(s):
  if not(s.startswith("\"") and s.endswith("\"")):
    return "".join(["\"", s, "\""])

def indent(level):
  return ("." * level * 2)

def main():
  print "webpage: {"
  level = 1
  pool = ConnectionPool("webpage")
  f = pycassa.ColumnFamily(pool, "f")
  for fk, fv in f.get_range(start="", finish=""):
    print indent(level), "key:", quote(fk), ","
    print indent(level), "f: {"
    if type(fv) == type(OrderedDict()):
      print_map(level+1, fv)
    else:
      print indent(level+1), fk, ":", quote(fv), ","
    print indent(level), "},"
    p = pycassa.ColumnFamily(pool, "p")
    print indent(level), "p: {"
    for pk, pv in p.get_range(start=fk, finish=fk):
      if type(pv) == type(OrderedDict()):
        print_map(level+1, pv)
      else:
        print indent(level+1), pk, ":", quote(pv), ","
    print indent(level), "},"
    sc = pycassa.ColumnFamily(pool, "sc")
    print indent(level), "sc: {"
    for sck, scv in sc.get_range(start=fk, finish=fk):
      if type(scv) == type(OrderedDict()):
        print_map(level+1, scv)
      else:
        print indent(level+1), sck, ":", quote(scv), ","
    print indent(level), "}"
  print "}"

if __name__ == "__main__":
  main()

The count_pages.py script lists all the records (by key) from the named column family (so you will have to pass in "f" or "p" as a parameter.

#!/usr/bin/python

import pycassa
from pycassa.pool import ConnectionPool
from pycassa.util import OrderedDict
import sys

def main():
  if len(sys.argv) != 2:
    print "Usage: %s col_fam_name" % (sys.argv[0])
    sys.exit(-1)
  col_fam = sys.argv[1]
  pool = ConnectionPool("webpage")
  cf = pycassa.ColumnFamily(pool, col_fam)
  res = cf.get_range(start="", finish="")
  count = 0
  for k, v in res:
    print k
    count = count + 1
  print "#-records in %s: %d\n" % (col_fam, count)

if __name__ == "__main__":
  main()

I am still learning Pycassa and Cassandra, so there is nothing fancy here - both scripts are fairly basic usages of the Pycassa API (more details can be found on the Pycassa Tutorial if you are interested).

Problems to Fix

While I am still convinced that the Nutch/GORA with Cassandra approach would be a good fit for my pipeline application, I did find a number of things that I don't quite like. Yeah, I know, picky, picky... :-).

First, I would like to restrict the crawl to the domain provided via the seed URL. I noticed that at the end of the third iteration, a lot of the records were for pages outside the parathyroid.com domain I specified in the seed URL - normally this would be the cue to stop crawling further. This works fine for discovery-style crawls, but for focused single-site crawls (which would be the kind of web crawls I envision doing more frequently), it would be good to be restrict the crawl to the domain specified, so successive iterations converge. This can be done by ensuring that anything in the webpage["sc"]["ol"] column gets promoted to the fetch list only if it matches the domain of its row key.

Second, the u_idx parameter I specified in the seed URL as metadata only gets written into the webpage record for that page. My intent in providing this metadata is to have it propagate through to all its children. I believe this can be done using a combination of fetch and parse filters.

Finally, I see that my u_idx metadata parameter doesn't make it to Solr at all (not even for the single record where this was added by Nutch). Since the whole point of adding metadata along with the seed URLs is so I can search indexes for those pages using the metadata, this kind of defeats the purpose. I think that the field just needs to be added to the solrindex-mapping.xml file in Nutch's runtime/local/config directory.

My next post would probably focus on solutions to these problems...

Solr Report Generation with Python, SimpleJson and GNU Parallel

2011-12-30T13:39:00.000-08:00

Recently, I needed to find if some fields were being correctly populated in our Solr index. To do this sort of ad-hoc reporting in the past (we used to be a Lucene shop), I would just write a simple Python/PyLucene script (or more recently a Jython script with embedded Java-Lucene calls or just a JUnit test which I could run from the command line with Ant), hop on to the machine hosting the index and run it. In our brave new Solr world, however, everything is available over HTTP, so I decided to see if I could do something similar over HTTP.

To provide a little background, the records in the index are book chapters. Chapters of the same book share some book related metadata, such as ISBN, which are denormalized into the chapter records. The objective was to see if two new such metadata fields (call them "meta1" and "meta2" for this discussion) was being populated correctly. The problem was that these were being provided from a separate (manually maintained) data source, so there was a chance that the coverage may not have been complete.

The Solr-Python wiki page lists some Solr clients for Python, but Solr also provides a JSON response writer, so as the wiki page mentions, one can just use simplejson library to read Solr's JSON output. This is what I did, since I did not want to (unnecessarily) commit to a specific Python/Solr API.

My first version made a call to find all the book chapter records to find the count of the books, then find the number of pages that I would need to loop through, then iterate through the pages, accumulating the META-1 and META-2 values in a pair of Python dictionaries keyed by ISBN. Once done, the script simply loops through the keys and prints out the values for unique META-1 and META-2, finally reporting the number of books where these fields did not get assigned. Here is the code.

# Source: src/scripts/myclient.py

from urllib2 import *
import simplejson
import urllib

# get count for query
server = "http://mysolr.hostname.com:8963/solr/select"
params = urllib.urlencode({
  "q" : "+contenttype:BOOK",
  "rows" : "1",
  "wt" : "json"
})
conn = urllib.urlopen(server, params)
rsp = simplejson.load(conn)
numfound = rsp["response"]["numFound"]
conn.close()
print numfound

# calculate the number of pages to iterate
rows_per_page = 25
num_pages = (numfound / rows_per_page) + (numfound % rows_per_page)

# iterate through the pages, accumulating data in dictionaries
isbn_meta1s = dict()
isbn_meta2s = dict()
for i in range(num_pages):
  print "processing page %d/%d" % (i, num_pages)
  params = urllib.urlencode({
    "q" : "+contenttype:BOOK",
    "start" : str(i * rows_per_page),
    "rows" : str(rows_per_page),
    "fl" : "isbn,meta1,meta2",
    "wt" : "json"
  })
  conn = urllib.urlopen(server, params)
  rsp = simplejson.load(conn)
  for doc in rsp["response"]["docs"]:
    try:
      (meta1) = doc["meta1"]
    except KeyError:
      meta1 = "999999"
    try:
      (meta2) = doc["meta2"]
    except KeyError:
      meta2 = "999999"
    isbn = doc["isbn"]
    isbn_meta1s[isbn] = meta1
    isbn_meta2s[isbn] = meta2
  conn.close()

# report
fout = open("/tmp/book_missing_metas.txt", "w")
fout.write("#" + "|".join(["ISBN", "META-1", "META-2"]) + "\n")
num_bad_meta1 = 0
num_bad_meta2 = 0
num_isbns = len(isbn_meta1s)
for isbn in isbn_meta1s.keys():
  meta1 = isbn_meta1s[isbn]
  if meta1 == "999999":
    num_bad_meta1 = num_bad_meta1 + 1
  meta2 = isbn_meta2s[isbn]
  if meta2 == "999999":
    num_bad_meta2 = num_bad_meta2 + 1
  fout.write("%s|%s|%s\n" % (isbn, meta1, meta1))

# stats
fout.write("# --\n")
fout.write("# bad meta1 = %d/%d, bad meta2 = %d/%d" % \
    (num_bad_meta1, num_isbns, num_bad_meta2, num_isbns))
fout.close()

To run it, we simply do something like this:

1	[spal@lysdexic src]$ python myclient.py

You could, of course, pass in a rows parameter equal to the response@numFound value and dispense with all the iterating, but I did not want to place too much load on the server (materializing large result sets requires more memory). The code above simulates a single user scrolling through the pages one by one, 25 records at a time, collecting data as it goes. The reporting "user" does not place too much strain on the Solr server, but it does take a while to complete if the number of pages are large (which it is in my case).

So I thought of parallelizing the task of hitting Solr using GNU Parallel - this places a little more load on the Solr server, but still very tolerable - instead of a single client, I decided to run 8 parallel clients. Plus, it helps me get my job done faster.

To make this code work with GNU Parallel, I had to split the processing up into three parts - the first part does the initial Solr call to get the count and calculates the number of pages, then writes the page numbers, one per line to STDOUT. The output of this is piped to the second part, which takes a page number as a command line argument and produces a list of pipe-separated values for ISBN, META-1 and META-2 fields. This output is piped to the third part, which accumulates the data into a dictionary and prints out the final report. Kind of similar to modeling a job as a map-reduce job. The three stages are shown below, they are mostly similar to the monolithic script shown above.

# Source: src/scripts/myclient-1.py

from urllib2 import *
import simplejson
import urllib

# get count for query
server = "http://mysolr.hostname.com:8963/solr/select"
params = urllib.urlencode({
  "q" : "+contenttype:BOOK",
  "rows" : "1",
  "wt" : "json"
})
conn = urllib.urlopen(server, params)
rsp = simplejson.load(conn)
numfound = rsp["response"]["numFound"]
rows_per_page = 25
num_pages = (numfound / rows_per_page) + (numfound % rows_per_page)
for pg in range(num_pages):
  print pg * rows_per_page

# Source: src/scripts/myclient-2.py

from urllib2 import *
import simplejson
import urllib
import sys

start = sys.argv[1]
rows_per_page = 25
server = "http://mysolr.hostname.com:8963/solr/select"
params = urllib.urlencode({
  "q" : "+contenttype:BOOK",
  "start" : str(start),
  "rows" : str(rows_per_page),
  "fl" : "isbn,meta1,meta2",
  "wt" : "json"
})
conn = urllib.urlopen(server, params)
rsp = simplejson.load(conn)
for doc in rsp["response"]["docs"]:
  try:
    (meta1) = doc["meta1"]
  except KeyError:
    meta1 = "999999"
  try:
    (meta2) = doc["meta2"]
  except KeyError:
    meta2 = "999999"
  print "|".join([doc["isbn"], meta1, meta2])
conn.close()

# Source: src/scripts/myclient-3.py

import sys

uniques = dict()
for line in sys.stdin:
  (isbn, meta1, meta2) = line[:-1].split("|")
  uniques[isbn] = "|".join([meta1, meta2])

fout = open("/tmp/book_missing_metas.txt", "w")
num_bad_meta1 = 0
num_bad_meta2 = 0
num_isbns = len(uniques)
for isbn in uniques.keys():
  (meta1, meta2) = uniques[isbn].split("|")
  if meta1 == "999999":
    num_bad_meta1 = num_bad_meta1 + 1
  if meta2 == "999999":
    num_bad_meta2 = num_bad_meta2 + 1
  fout.write("%s|%s|%s\n" % (isbn, meta1, meta2))
# statistics
fout.write("----\n")
fout.write("Bad CIDs: %d/%d, Bad PIE_CIDs: %d/%d\n" % \
    (num_bad_meta1, num_isbns, num_bad_meta2, num_isbns))
fout.close()

To run this job with parallel with 8 clients calling Solr (the number of CPUs on my desktop, although the gating factor is really the number of simultaneous requests that the Solr server can handle without too much latency), we use the following command:

1 2	[spal@lysdexic src]$ python myclient-1.py \| \ parallel -P 8 "python myclient-2.py {}" \| python myclient-3.py

As expected, this works much faster.