Showing posts with label xml. Show all posts
Showing posts with label xml. Show all posts

Saturday, January 13, 2018

Crowdsourcing a Labeling task using Amazon Mechanical Turk


Happy New Year! My New Year's resolution for 2018 is, perhaps unsurprisingly, to blog more frequently than I have in 2017.

Despite the recent advances in unsupervised and reinforcement learning, supervised learning remains the most time-tested and reliable method to build Machine Learning (ML) models today, as long as you have enough training data. Among ML models, Deep Learning (DL) has proven to be more effective in many cases. DL's grreatest advantage is its capability to consider all sorts of non-linear feature interactions automatically. In return all it asks for is more processing power and more training data.

With the ubiquity of the computer and the Internet in our everyday lives, it is not surprising that our very act of collective living generates vast amounts of data. In many cases it is possible, with a little bit of ingenuity, to discover implicit labels in this data, making the data usable for training supervised DL models. In most other cases, we are not so lucky and must take explicit steps to generate these labels. Traditionally, people have engaged human experts to do tha labeling from scratch, but this is usually very expensive and time-consuming. More recently, the trend is to generate noisy labels using unsupervised techniques, and validate them using human feedback.

Which brings me to the subject of my current post. One way to get this human feedback is through Amazon's Mechanical Turk (AMT or MTurk), where you can post a Human Intelligence Task (HIT) and have people do these HITs in return for micropayments made through the MTurk network. In this post, I describe the process of creating a collection of HITs and making them available for MTurk workers (aka turkers), then collecting the resulting labels.

Problem Description


I was trying to generate tags for snippets of text. These tags are intended to be keywords that are self-contained and describe some aspect of the text. And yes, I realize that this looks like something plain old search could do as well, but bear with me here -- this data is a first step of a larger pipeline and I do need these multi-word labels.

So each record consists of a snippet of text and 5 multi-word candidate labels. The labels are generated using various unsupervised techniques, some rule based and some that exploit statistical features of language. Because the scoring is not compatible across the various techniques, we select the top 10 percentile from each set, then randomly chose 5 labels for each snippet from the merged label pool.

The first step is to pre-pay for the HITs and push them to the MTurk site where they become visible to turkers, some of whom will take them on and complete them. After the specified number of turkers have completed the HITs to assign their crowdsourced labels and we accept their work, they get paid by AMR, and we need to download their work. MTurk provides an API that allows you to upload the HITs and retrieve the crowdsourced labels, which I will talk about here. My coverage is more from a programming standpoint, so I have done this against the MTurk sandbox site, which is free to use.

In terms of required software, I recently upgraded to Anaconda Python3. The other libraries used are boto3 to handle the network connections, the jinja2 templating engine included with Anaconda for generating the XML for the HIT in the MTurk request, and xmltodict to parse XML payloads in the MTurk response to Python data structures. Both boto3 and xmltodict can be installed using pip install. I also had a lot of help from this post Tutorial: A beginner's guide to crowdsourcing ML training data with Python and MTurk on the MTurk blog.

Creating HITs and uploading to MTurk


The unsupervised algorithms are run and the top results from each merged on our Apache Spark based analytics platform. A sample of these merged results are downloaded and used as input for creating the HITs. The input data looks like this:


The first step is to establish a connection to the MTurk (sandbox) server. For this, you need to have an AWS account, an MTurk development/requester account, and also link your AWS account to the MTurk account. This AWS Documentation page covers these steps in more detail. Once you are done, you should be able to establish a connection to the sandbox and see how much pretend money you have in the sandbox to pay your pretend workers.

1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from jinja2 import Template
import boto3
import os

# constants
MTURK_SANDBOX = "https://mturk-requester-sandbox.us-east-1.amazonaws.com"
MTURK_REGION = "us-east-1"
MTURK_PREVIEW_URL = "https://workersandbox.mturk.com/mturk/preview?groupId={:s}"

DATA_DIR = "../data"
HIT_ID_FILE = os.path.join(DATA_DIR, "best-keywords-hitids.txt")

NUM_QUESTIONS_PER_HIT = 10

# extract AWS credentials from local file
creds = []
CREDENTIALS_FILE = "/path/to/amazon-credentials.txt"
with open(CREDENTIALS_FILE, "r") as f:
    for line in f:
        if line.startswith("#"):
            continue
        _, cred = line.strip().split("=")
        creds.append(cred)

# verify that we can access MTurk sandbox server
mturk = boto3.client('mturk',
   aws_access_key_id=creds[0],
   aws_secret_access_key=creds[1],
   region_name=MTURK_REGION,
   endpoint_url=MTURK_SANDBOX
)
print("Sandbox account pretend balance: ${:s}".format(
    mturk.get_account_balance()["AvailableBalance"]))

We have (in our example) just 24 snippets with associated keywords. I want to group them into 10 snippets per HIT, so I have 3 HITs with 10, 10 and 4 snippets respectively. In reality you want a larger number for labeling, but since I was in development mode, I was the person doing the HIT each time, and I wanted to minimize my effort. At the same time, I wanted to make sure I could group my input into HITs of 10 snippets each, hence the choice of 24 snippets.

Each HIT needs to get formatted as an HTML form, which is then embedded inside a HTMLQuestion tag that is part of the XML syntax MTurk understands. Since we wanted to put multiple snippets into a single HIT, it was more convenient to use the loop unrolling capabilities of the Jinja2 templating engine than rely on Python's native templating through format() calls. Here is the template for our HIT.

1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
full_xml = Template("""
<HTMLQuestion xmlns="http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2011-11-11/HTMLQuestion.xsd">
    <HTMLContent><![CDATA[
<!DOCTYPE html>
<html>
    <head>
        <meta http-equiv='Content-Type' content='text/html; charset=UTF-8'/>
        <script type='text/javascript' src='https://s3.amazonaws.com/mturk-public/externalHIT_v1.js'></script>
    </head>
    <body>
        <form name="mturk_form" method="post" id="mturk_form" 
              action="https://www.mturk.com/mturk/externalSubmit">
        <input type="hidden" value="" name="assignmentId" id="assignmentId" />
        <ol>
        {% for row in rows %}
            <input type="hidden" name="iid_{{ row.id }}" value="{{ row.iid }}"/>
            <li>
                <b>Select all keywords appropriate for the snippet below:</b><br/>
                {{ row.snippet }}
                <p>
                <input type="checkbox" name="k_{{ row.id }}_1">{{ row.keyword_1 }}<br/>
                <input type="checkbox" name="k_{{ row.id }}_2">{{ row.keyword_2 }}<br/>
                <input type="checkbox" name="k_{{ row.id }}_3">{{ row.keyword_3 }}<br/>
                <input type="checkbox" name="k_{{ row.id }}_4">{{ row.keyword_4 }}<br/>
                <input type="checkbox" name="k_{{ row.id }}_5">{{ row.keyword_5 }}<br/>
                </p>
            </li>
            <hr/>
        {% endfor %}
        </ol>

            <p><input type="submit" id="submitButton" value="Submit"/>
            </p>
        </form>
        <script language='Javascript'>turkSetAssignmentID();</script>
    </body>
</html>
]]>
    </HTMLContent>
    <FrameHeight>600</FrameHeight>
</HTMLQuestion>
""")

We then group our data into 10 rows each, create a data structure rows, each row of which contains a dictionary of field names and values, then render the snippet above for this data structure. The resulting XML is fed to the MTurk sandbox server using boto3. Each call corresponds to a single HIT and the server will return a corresponding HIT Id, which we save for later use. It also returns a HIT group ID which we will use to generate a set of preview URLs.

We have modeled each group of 10 snippets as a completely separate HITs, with its own unique title (trailing #n). We could also have run multiple create_hit calls using the same title, in which case, a group of HITs are created under the same title. However, I noticed that I was sometimes getting back duplicate HIT Ids in that case, so I went with the separate HIT per 10 snippets strategy.

I also found a good use for the Keywords parameter - if you put some oddball term in there, you could share it with your team to get back the list of HITs you want them to look at.

1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def create_hit(mturk, question, hit_seq):
    hit = mturk.create_hit(
        Title="Best Keywords in Caption #{:d}".format(hit_seq),
        Description="Find best keywords in caption text",
        Keywords="aardvaark",
        Reward="0.10",
        MaxAssignments=1,
        LifetimeInSeconds=172800,
        AssignmentDurationInSeconds=600,
        AutoApprovalDelayInSeconds=14400,
        Question=question
    )
    group_id = hit["HIT"]["HITGroupId"]
    hit_id = hit["HIT"]["HITId"]
    return group_id, hit_id


rows = []
hit_group_ids, hit_ids = [], []
hit_seq = 1
with open(os.path.join(DATA_DIR, "best-keywords.tsv"), "r") as f:
    for lid, line in enumerate(f):
        if lid > 0 and lid % NUM_QUESTIONS_PER_HIT == 0:
            question = full_xml.render(rows=rows)
            hit_group_id, hit_id = create_hit(mturk, question, hit_seq)
            hit_group_ids.append(hit_group_id)
            hit_ids.append(hit_id)
            rows = []
            hit_seq += 1
        iid, snippet, key_1, key_2, key_3, key_4, key_5 = line.strip().split("\t")
        row = {
            "id": (lid + 1),
            "iid": iid, 
            "snippet": snippet,
            "keyword_1": key_1,
            "keyword_2": key_2,
            "keyword_3": key_3,
            "keyword_4": key_4,
            "keyword_5": key_5,
        }
        rows.append(row)
        
if len(rows) > 0:
    question = full_xml.render(rows=rows)
    create_hit(mturk, question, hit_seq)
    hit_group_ids.append(hit_group_id)
    hit_ids.append(hit_id)

The code above results in a flat file of HIT Ids that I can use to recall results for these HITs later. You can also see your HITs appear as shown below:


As you might expect, this is a giant form consisting of text snippets separated by checkbox group of 5 candidate keywords, terminated with a single Submit button. I am not sure if you can have Javascript support for more sophisticated use cases, but you can do a lot with HTML5 nowadays. Here is what (part of) the form looks like, marked up by the dev turker (me :-)).



Retrieving crowdsourced labels on HITs from MTurk


In a real-life scenario, the HITs would be on the MTurk production server and real humans would (hopefully) find my micro-payment of 10 cents per HIT adequate and do the marking up for me. I have configured my HIT to have MaxAssignments=1, which means I want only 1 worker to work on the HIT -- in reality, you want at least 3 people to work on each HIT so you can do a majority vote (or something more sophisticated) on their labels. In any case, once all your HITs have been handled by the required number of turkers, it is time to download the results.

Results for a HIT can be retrieved using the list_assignments_for_hit() method of the MTurk client -- you need the HIT Id for the HIT that was returned during HIT creation, and which we had stored away for use now. The response from the MTurk server is a JSON response, with the actual Answer value packaged as an XML payload. We use the xmltodict.parse() method to parse this payload into a Python data structure, which we then pick apart to write out the output.

1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import boto3
import os
import xmltodict

# constants
MTURK_SANDBOX = "https://mturk-requester-sandbox.us-east-1.amazonaws.com"
MTURK_REGION = "us-east-1"

DATA_DIR = "../data"
HIT_ID_FILE = os.path.join(DATA_DIR, "best-keywords-hitids.txt")
RESULTS_FILE = os.path.join(DATA_DIR, "best-keywords-results.txt")

# extract AWS credentials from local file
creds = []
CREDENTIALS_FILE = "/path/to/amazon-credentials.txt"
with open(CREDENTIALS_FILE, "r") as f:
    for line in f:
        if line.startswith("#"):
            continue
        _, cred = line.strip().split("=")
        creds.append(cred)

# verify access to MTurk
mturk = boto3.client('mturk',
   aws_access_key_id=creds[0],
   aws_secret_access_key=creds[1],
   region_name=MTURK_REGION,
   endpoint_url=MTURK_SANDBOX
)
print("Sandbox account pretend balance: ${:s}".format(
    mturk.get_account_balance()["AvailableBalance"]))

# get HIT Ids stored from during HIT creation
hit_ids = []
with open(HIT_ID_FILE, "r") as f:
    for line in f:
        hit_ids.append(line.strip())

# retrieve MTurk results
fres = open(RESULTS_FILE, "w")
for hit_id in hit_ids:
    snippet_ids, keyword_ids = {}, {}
    results = mturk.list_assignments_for_hit(HITId=hit_id, 
        AssignmentStatuses=['Submitted'])
    if results["NumResults"] > 0:
        for assignment in results["Assignments"]:
            worker_id = assignment["WorkerId"]
            answer_dict = xmltodict.parse(assignment["Answer"])
            answer_dict_2 = answer_dict["QuestionFormAnswers"]["Answer"]
            for answer_pair in answer_dict_2:
                field_name = answer_pair["QuestionIdentifier"]
                field_value = answer_pair["FreeText"]
                if field_name.startswith("iid_"):
                    id = field_name.split("_")[1]
                    snippet_ids[id] = field_value
                    keyword_ids[id] = []
                else:
                    _, iid, kid = field_name.split("_")
                    keyword_ids[iid].append(kid)
    for id, iid in snippet_ids.items():
        selected_kids = ",".join(keyword_ids[id])
        fres.write("{:s}\t{:s}\t{:s}\n".format(worker_id, iid, selected_kids))

fres.close()

The output of this step is a TSV file that contains the worker ID, the snippet ID, and a comma-separated list of keyword IDs that were found to be meaningful by the turker(s). This can now be joined with the original input file to find the preferred labels.

1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
A2AQYARTZTL5EE S0735109710021418-gr5 2,3
A2AQYARTZTL5EE S0894731707005962-gr1 2,4
A2AQYARTZTL5EE S1740677311000118-gr2 1,5
A2AQYARTZTL5EE S0031938414005393-gr2 1,3,4,5
A2AQYARTZTL5EE S1542356515000415-gr2 3
A2AQYARTZTL5EE S1521661616300158-gr8 2
A2AQYARTZTL5EE S0091743514001212-gr2 1,2,3
A2AQYARTZTL5EE S0735109712023662-gr2 1,2,3
A2AQYARTZTL5EE S0026049509000456-gr1 
A2AQYARTZTL5EE S0079610715000103-gr3 1,3
...

This is all I have for today. I hope you enjoyed the post and found it useful. I believe crowdsourcing will become more important as people begin to realize the benefits of weak supervision, and the MTurk API makes it quite easy to set up this kind of jobs.


Friday, February 10, 2012

Nutch/GORA - Using a sitemap to seed a site

The genesis of this "feature" was due to my misunderstanding of how Nutch works. So far I've been running fairly small batches as I built my plugin and other application specific custom code. But I was now at the point where I could try ingesting all the XML files provided by this provider, so I did. There are about 6000 XML files in this collection, but Nutch fetched exactly 318. Every time (I tried it couple of times to make sure I was doing it right).

I initially thought that perhaps because the seed list for the provider was in HTML, Nutch's default HTML parser was doing some magic "above the fold" scoring that discounted items further down the page, so I hit upon the idea of using a sitemap XML file. I figured that since Nutch didn't provide sitemap support, I'd have to write my own parser (which wouldn't have any magic scoring). Since my XML parser plugin already allowed for multiple parsers, this just involves writing a sitemap XML processor and calling it through my plugin.

Of course, this did not fix the problem, the fetcher just stopped after a different number of files. Turns out that by default, Nutch only reads the first 64KB of the file and drops the rest. A quick peek at my webpage["f"]["cnt"] in the database confirmed this. So the fix for my original problem was really just adding this block to my nutch-site.xml file:

1
2
3
4
5
6
7
8
9
<property>
  <name>http.content.limit</name>
  <value>-1</value>
  <description>The length limit for downloaded content using the http
  protocol, in bytes. If this value is nonnegative (>=0), content longer
  than it will be truncated; otherwise, no truncation at all. Do not
  confuse this setting with the file.content.limit setting.
  </description>
</property>

But I had already written the sitemap parser, and using a seed file in sitemap format would allow me to also support vertical crawls for partner sites (who typically provide us a URL to their sitemap) with the same infrastructure that I am building for ingesting provider XML files. So I decided to go with it.

The sitemap format is quite simple, but the only information thats usable during the inject stage is the urlset/url/loc value. The provider_index method in the updated CherryPy server code below generates a dynamic sitemap XML from the contents of the filesystem.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
#!/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/directory"
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 = []
    name = name.replace("-sitemap.xml", "")
    os.path.walk(os.path.join(SITES_DIR, "providers", name), \
      _accumulate_files, files)
    index = """<?xml version=\"1.0\" encoding=\"UTF-8\"?>
<urlset xmlns=\"http://www.sitemaps.org/schemas/sitemap/0.9\">
    """
    for file in files:
      url = "http://%s:%s/provider/%s" % (SERVER_HOST, SERVER_PORT, \
        urllib.quote_plus(file.replace(os.path.sep, "__")))
      index += """
<url><loc>%s</loc></url>
      """ % (url)
    index += """
</urlset>
    """
    cherrypy.response.headers["Content-Type"] = "application/xml"
    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(), '/')

The ProviderXmlProcessorFactory (described in a previous post) was modified slightly to check for the basename of the URL. If the basename contains the string "sitemap", then it will delegate to the Sitemap Processing component first, and only then look at the value of the u_idx metadata field. The code for this is shown below:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
// Source: src/plugin/mycompany/src/java/com/mycompany/nutch/parse/xml/sitemap/SitemapXmlProcessor.java
package com.mycompany.nutch.parse.xml.sitemap;

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

import org.codehaus.jackson.map.ObjectMapper;
import org.jdom.Document;
import org.jdom.Element;
import org.jdom.Namespace;
import org.jdom.input.SAXBuilder;
import org.xml.sax.InputSource;

import com.mycompany.nutch.parse.xml.IProviderXmlProcessor;
import com.mycompany.nutch.parse.xml.ProviderXmlFields;

public class SitemapXmlProcessor implements IProviderXmlProcessor {

  public static final String OUTLINKS_KEY = "_outlinks";
  
  private ObjectMapper mapper;
  
  public SitemapXmlProcessor() {
    mapper = new ObjectMapper();
  }
  
  @SuppressWarnings("unchecked")
  @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();
    Namespace ns = root.getNamespace();
    if ("urlset".equals(root.getName())) {
      List<String> urls = new ArrayList<String>();
      List<Element> eUrls = root.getChildren("url", ns);
      for (Element eUrl : eUrls) {
        urls.add(eUrl.getChildTextTrim("loc", ns));
        // sitemap 0.9 also specifies optional elements lastmod,
        // changefreq and priority. The first two could be handled
        // if we change the OutLink object to hold these values
        // as metadata, which is used to update the modified
        // and fetchInterval values in the outlink once its put
        // into the fetchlist. But we ignore these currently.
      }
      parsedFields.put(OUTLINKS_KEY, mapper.writeValueAsString(urls));
      // set some fields to prevent nutch from choking on NPEs
      parsedFields.put(ProviderXmlFields.title.name(), "sitemap");
      parsedFields.put(ProviderXmlFields.content.name(), "sitemap");
    }
    return parsedFields;
  }
}

As you can see, it parses out the URLs and accumulates them into a List, whcih is then written out to the parsedFields map as a JSON string keyed by a magic key "_outlinks". The ProviderXmlParser plugin does a bit of special processing for this key, specifically it converts the JSON string back to a List and writes the list elements out to the webpage["f"]["ol"] column. Here is the modified ProviderXmlParser class.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
// 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.ArrayList;
import java.util.Collection;
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.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;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.type.TypeReference;

import com.mycompany.nutch.parse.xml.sitemap.SitemapXmlProcessor;

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;
  private ObjectMapper mapper;
  private TypeReference<List<String>> outlinksTypeRef;
  
  public ProviderXmlParser() {
    this.mapper = new ObjectMapper();
    this.outlinksTypeRef = new TypeReference<List<String>>() {};
  }
  
  @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(url, 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) ||
                SitemapXmlProcessor.OUTLINKS_KEY.equals(key)) {
              continue;
            }
            page.putToMetadata(new Utf8(key), 
              ByteBuffer.wrap(parsedFields.get(key).getBytes()));
          }
          if (parsedFields.containsKey(
              SitemapXmlProcessor.OUTLINKS_KEY)) {
            // if we have OUTLINKS data, then populate it as well
            List<String> outlinkUrls = mapper.readValue(
              parsedFields.get(SitemapXmlProcessor.OUTLINKS_KEY), 
              outlinksTypeRef);
            Outlink[] outlinks = new Outlink[outlinkUrls.size()];
            for (int i = 0; i < outlinks.length; i++) {
              String outlinkUrl = outlinkUrls.get(i);
              outlinks[i] = new Outlink(outlinkUrl, outlinkUrl);
            }
            parse.setOutlinks(outlinks);
          }
          parse.getParseStatus().setMajorCode(ParseStatusCodes.SUCCESS);
        } catch (Exception e) {
          LOG.warn("Parse of URL: " + url + " failed", e);
          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;
  }
}

And thats pretty much it. We now inject the following seed URL for this provider, and run through two iterations (depth 2) of the generate, fetch, parse and updatedb, and as expected, all the provider XML files are ingested without problems.

1
http://localhost:8080/provider_index/prov1-sitemap.xml u_idx=prov1

The approach I describe is probably not what you would normally think of when you hear "nutch" and "sitemap enabled" in the same sentence. After all, we are throwing away the optional metadata that is being provided to us, such as crawl frequency and last modified time. Unfortunately with the approach I have chosen - using the sitemap XML file as the seed URL - the only way I know of to ingest this information in a single pass is to change the Outlink data structure. However, there is nothing preventing you from making a second pass over the sitemap after the fetch and then resetting the fetch interval for a page based on its sitemap properties - sort of like my Delta Indexer on auto-pilot.

Friday, January 20, 2012

Nutch/GORA - Parsing Custom XML

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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
#!/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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
<?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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
// 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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
// 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);
  }
}
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
// 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;

}
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
// 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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
// 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(" ");
  }
}
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
// 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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
<!-- 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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
// 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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
<?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).

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
<?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.