Showing posts with label json. Show all posts
Showing posts with label json. 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.


Sunday, December 29, 2013

Akka Content Ingestion Pipeline, Part III


In this post, I add a JSON/HTTP front end to my Akka Content Ingestion Pipeline. This allows clients remote access (over HTTP) to the pipeline, so they can submit jobs to it and make some rudimentary queries against it. As you already know, a client can send Fetch messages to the pipeline to have a document be crawled off a website, parsed and indexed into a Solr index, a Stats message to query the size of the pipeline's internal queues, and a Stop message to terminate the pipeline.

The front end is a HTTP server that listens on a specified host and port and forwards HTTP GET and PUT requests to an Actor adapted for listening to HTTP requests (via the HttpServiceActor mixin). The PUT requests are accompanied by JSON payloads which correspond to the data in the message case classes. The actor's receive() method responds to these requests by transforming the JSON to the equivalent message case classes and sending the message to the controller Actor.

In addition, the front end HTTP server has a shutdown hook that will terminate the Controller and its children in an orderly fashion (waiting until the queues are all drained) by sending it a Stop request.

In pictures, this work corresponds to the top block of the diagram (updated from last week) below:


To build the HTTP server, I used the Spray library, which allows you to build an HTTP server in just a few lines of Scala code:

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
// Source: src/main/scala/com/mycompany/delsym/rest/Main.scala
package com.mycompany.delsym.rest

import com.typesafe.config.ConfigFactory

import akka.actor.ActorSystem
import akka.actor.Props
import akka.actor.actorRef2Scala
import akka.io.IO
import spray.can.Http
import spray.httpx.RequestBuilding.Get

object Main extends App {

  implicit val system = ActorSystem("DelSym")
  
  val conf = ConfigFactory.load()
  val host = conf.getString("delsym.rest.host")
  val port = conf.getInt("delsym.rest.port")

  val api = system.actorOf(Props[RestActor], "api")
  IO(Http) ! Http.Bind(api, host, port = port)
  
  sys.addShutdownHook {
    Console.println("Shutting down...")
    api ! Get("/stop")
  }
}

The HTTP Server starts up a RestActor which is a specialized Actor (providing a HTTP Service). Its receive method does pattern matching on the requests and accordingly calls messages on the underlying Controller. The receive method is built off a routing table that is built using the Spray routing DSL. The code for the RestActor 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
92
93
94
95
96
97
// Source: src/main/scala/com/mycompany/delsym/rest/RestActor.scala
package com.mycompany.delsym.rest

import scala.concurrent.Await
import scala.concurrent.duration.DurationInt

import com.mycompany.delsym.actors.Controller
import com.mycompany.delsym.actors.Fetch
import com.mycompany.delsym.actors.Index
import com.mycompany.delsym.actors.MessageProtocol
import com.mycompany.delsym.actors.Parse
import com.mycompany.delsym.actors.Stats
import com.mycompany.delsym.actors.Stop
import com.typesafe.config.ConfigFactory

import akka.actor.Actor
import akka.actor.Props
import akka.actor.actorRef2Scala
import akka.pattern.ask
import akka.util.Timeout
import spray.httpx.SprayJsonSupport.sprayJsonUnmarshaller
import spray.httpx.marshalling.ToResponseMarshallable.isMarshallable
import spray.json.pimpAny
import spray.routing.Directive.pimpApply
import spray.routing.HttpService

class RestActor extends Actor with RestService {

  val conf = ConfigFactory.load()
  implicit val timeout = Timeout(
    conf.getInt("delsym.rest.timeout").seconds)

  val controller = actorRefFactory.actorOf(
    Props[Controller], "controller")

  def actorRefFactory = context
  
  def receive = runRoute {
    (get & path("stats")) {
      jsonpWithParameter("callback") {
        complete {
          val future = (controller ? Stats(Map.empty))
            .mapTo[Stats]
          val result = Await.result(future, timeout.duration)
          import MessageProtocol.statsFormat
          result.toJson.prettyPrint
        }
      }
    } ~
    (put & path("fetch")) { 
      jsonpWithParameter("callback") {
        import MessageProtocol.fetchFormat
        entity(as[Fetch]) { fetch => 
          complete {
            controller ! fetch
            "Got(" + fetch.toJson.compactPrint + ")"
          }  
        }
      }
    } ~
    (put & path("parse")) { 
      jsonpWithParameter("callback") {
        import MessageProtocol.parseFormat
        entity(as[Parse]) { parse => 
          complete {
            controller ! parse
            "Got(" + parse.toJson.compactPrint + ")"
          }  
        }
      }
    } ~
    (put & path("index")) { 
      jsonpWithParameter("callback") {
        import MessageProtocol.indexFormat
        entity(as[Index]) { index => 
          complete {
            controller ! index
            "Got(" + index.toJson.compactPrint + ")"
          }  
        }
      }
    } ~
    (get & path("stop")) { 
      complete {
        import MessageProtocol.stopFormat
        controller ! Stop(0)
        "Stop signal sent"
      }
    }    
  }
}

trait RestService extends HttpService {

  implicit def executionContext = 
    actorRefFactory.dispatcher
}

Spray also provides JSON marshalling/unmarshalling facilities. This is automatic for native types and collections, but for case classes, it is necessary to specify the protocol. Since our messages are all case classes, we specify the protocol as below. This protocol needs to be brought into scope just before the actual JSON marshalling/unmarshalling, which is why we have the import MessageProtocol.*Format calls in the code above.

1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
// Source: src/main/scala/com/mycompany/delsym/actors/DelsymMessage.scala
package com.mycompany.delsym.actors

import akka.actor.ActorRef
import spray.json._
import DefaultJsonProtocol._

...

/////////////// Message <--> JSON ser/deser ////////////

object MessageProtocol extends DefaultJsonProtocol {
  implicit val fetchFormat = jsonFormat3(Fetch)
  implicit val parseFormat = jsonFormat1(Parse)
  implicit val indexFormat = jsonFormat1(Index)
  implicit val statsFormat = jsonFormat1(Stats)
  implicit val stopFormat = jsonFormat1(Stop)
}

To test this, I used cURL to send in a stats GET request and a fetch PUT request. The commands and their outputs are shown below:

1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
sujit@tsunami:~/Projects/delsym$ curl localhost:8080/stats
{
  "stats": {
    "parsers": 0,
    "fetchers": 0,
    "indexers": 0
  }
}
sujit@tsunami:~/Projects/delsym$ curl -X PUT \
    -H "Content-Type: application/json \
    -d '{"url":"http://www.foo.com/bar", "depth":0, "metadata": {}}' \
    http://localhost:8080/fetch
Got({"url":"http://www.foo.com/bar","depth":0,"metadata":{}})

Although there is not that much code to show for it, this took me almost 2 days of experimentation to get right, mainly because Spray seems to make heavy use of implicits which are not really evident unless you read the documentation thoroughly. Here are some sites that helped me figuring things out.

  • GitHub Gist from Ayose Crzorla demonstrating a very simple Scala application that talks to two HttpServiceActors. This is what I started with.
  • Spray (REST on Akka) slides from Mathias Doenitz's talk in Paris Scala IO. This gave me some directive patterns that I used to implicitly convert JSON into message case classes and generate JSONP (although the wrapping in the callback doesn't work because I don't know how to set the content type).
  • Brandon Amos's Blog Post on adding shutdown hooks in Scala.

In addition to this, I also studied the code samples from the Akka in Action book, and downloaded the examples provided by Spray looking for useful patterns.

This part got done a bit earlier than planned, probably because I can sit around doing this at home all day over our Christmas to New Year office closure, but I am going to publish it anyway and move on to looking at how to distribute this application across multiple servers next. So in (the very likely) case that I don't post again before next year, belated Merry Christmas wishes and I hope you have a very Happy New Year and good times ahead in 2014.

Update 2014-01-01: For the remoting work, sbt gave me errors trying to download akka-remote for Akka 2.1.2 (and Spray 1.1-20130123) which I was working with so far (based on the version in the code for the Akka in Action book). So I upgraded Akka to the current latest stable version (Akka 2.2.3 and Spray 1.2.0) as a result of which both the classes in this post failed to compile. I had to rewrite them against the new API (using code examples from spray-routing examples). I have updated the code in the post to match the one in the DelSym GitHub repo.

Friday, November 16, 2012

An ElasticSearch Web Client with Scala and Play2

In this post, I describe the second part of my submission for the Typesafe Developer Contest. This part is a rudimentary web based search client to query an ElasticSearch (ES) server. It is a Play2/Scala web application that communicates with the ES server via its JSON query DSL.

The webapp has a single form that allows you to specify a Lucene query and various parameters and returns a HTML or JSON response. It will probably remind Solr developers of the admin form. I find the Solr admin form very useful for trying out qeries before baking them into code, and I envision a similar use for this webapp for ES search developers.

Since ES provides a rich JSON based Query DSL, the form here has a few more features than the Solr admin form, such as allowing for faceting and sorting. Although in the interests of full disclosure, it provides only a subset of the variations possible via direct use of JSON and curl on the command line. But its good for quick and dirty verification of search ideas. In order to quickly get started with ES's query DSL, I found this DZone article by Peter Kar and this blog post by Pulkit Singhal very useful (apart from the ES docs themselves, of course).

Since Play2 was completely new to me a week ago and now I am the proud author of a working webapp, I would like to share with you some of my insights into this framework. I typically learn new things by making analogies to stuff I already know, so I will explain Play2 by making analogies to Spring. If you know Spring, it may be helpful, and if you don't, well, maybe it was not that terribly helpful anyway...

Routing in Play2 is done using the conf/routes file, which maps URL patterns and HTTP methods to Play2 controller actions. Actions can be thought of as @RequestMapping methods in a Multi-action Spring controller, and are basically functions that transform a Request into a Response. A response can be a String wrapped in an Ok() method or it can be a method call into a view with some data, which returns a templated string to Ok(). There, thats it - about everything you need to know about Play2 to get to using it.

Unlike the last time (with Akka), this time around I did not use the Typesafe Play tutorial. Instead I downloaded Play2 and used the play command to build a new web application template (play new search), then to compile and run it. The best tutorial I found was this one on flurdy.com, which covers everything from choice of IDE to deployment on Heroku and everything in between. Other useful sources are Play's documentation (available with the Play2 download) and this example Play2 app on GitHub.

Here is my conf/routes file. I added the two entries under Search pages. They both respond to HTTP GET requests and call the form() and search() Actions respectively. The other two entries come with the generated project and are needed (so don't delete them).

1
 2
 3
 4
 5
 6
 7
 8
 9
10
# conf/routes
# Home page
GET     /                           controllers.Application.index

# Search pages
GET     /form                       controllers.Application.form
GET     /search                     controllers.Application.search

# Map static resources from the /public folder to the /assets URL path
GET     /assets/*file               controllers.Assets.at(path="/public", file)

There is another file in the conf directory, called conf/application.conf. It contains properties required by the default application. I added a new property for the URL for the ES server in this file.

1
2
3
# conf/application.conf
...
es.server="http://localhost:9200/"

The Play2 "new" command also generates a skeleton controller app/controllers/Application.scala, into which we add the two new form and search Actions. Here is the completed Application.scala file.

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
// app/controllers/Application.scala
package controllers

import models.{Searcher, SearchParams}
import play.api.data.Forms.{text, number, mapping}
import play.api.data.Form
import play.api.libs.json.{Json, JsValue}
import play.api.libs.ws.WS
import play.api.mvc.{Controller, Action}
import play.api.Play

object Application extends Controller {

  // define the search form
  val searchForm = Form(
    mapping (
      "index" -> text,
      "query" -> text,
      "filter" -> text,
      "start" -> number,
      "rows" -> number,
      "sort" -> text,
      "writertype" -> text,
      "fieldlist" -> text,
      "highlightfields" -> text,
      "facetfields" -> text
    ) (SearchParams.apply)(SearchParams.unapply)
  )
  
  // configuration parameters from conf/application.conf
  val conf = Play.current.configuration
  val server = conf.getString("es.server").get

  // home page - redirects to search form
  def index = Action {
    Redirect(routes.Application.form)
  }

  // form page
  def form = Action {
    val rsp = Json.parse(WS.url(server + "_status").
      get.value.get.body)
    val indices = ((rsp \\ "indices")).
      map(_.as[Map[String,JsValue]].keySet.head)
    Ok(views.html.index(indices, searchForm))
  } 

  // search results action - can send view to one of
  // three different pages (xmlSearch, jsonSearch or htmlSearch)
  // depending on value of writertype
  def search = Action {request =>
    val params = request.queryString.
      map(elem => elem._1 -> elem._2.headOption.getOrElse(""))
    val searchParams = searchForm.bind(params).get
    val result = Searcher.search(server, searchParams)
    searchParams.writertype match {
      case "json" => Ok(result.raw).as("text/javascript")
      case "html" => Ok(views.html.search(result)).as("text/html")
    }
  }
}

We first define a Search form and map it to the SearchParams class (defined in the model, below). The index Action has been changed to redirect to the form Action. The form method makes a call to the ES server to get a list of indexes (ES can support multiple indexes with different schemas within the same server), and then delegates to the index view with this list and an empty searchForm.

The search Action binds the request to the searchParams bean, then sends this bean to the Searcher.search() method, which returns a SearchResult object containing the results of the search. Two different views are supported - the HTML view (delegating to the search view template) and the raw JSON view that just dumps the JSON response from ES.

The respective views for the form and search are shown below. Not much to explain here, except that its another templating language that you have to learn. Its set up like a function - you pass in parameters that you use in the template. I followed the lead of the flurdy.com tutorial referenced above and kept it as HTML-ish as possibly, but Play2 has an extensive templating language of its own that you may prefer.

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
@** app/views/index.scala.html **@
@(indices: Seq[String], searchForm: Form[SearchParams])

@import helper._

@main("Search with ElasticSearch") {
  
  <h2>Search with ElasticSearch</h2>
  @form(action = routes.Application.search) {  
    <fieldset>
      <legend>Index Name</legend>
      <select name="index">
      @for(index <- indices) {
        <option value="@index">@index</option>
      }
      </select>
    </fieldset>
    <fieldset>
      <legend>Lucene Query</legend>
      <input type="textarea" name="query" value="*:*" maxlength="1024" rows="10" cols="80"/>
    </fieldset>
    <fieldset>
      <legend>Filter Query</legend>
      <input type="textarea" name="filter" value="" maxlength="512" rows="5" cols="80"/>
    </fieldset>  
    <fieldset>
      <legend>Start Row</legend>
      <input type="text" name="start" value="0" maxlength="5"/>
    </fieldset>
    <fieldset>
      <legend>Maximum Rows Returned</legend>
      <input type="text" name="rows" value="10" maxlength="5"/>
    </fieldset>
    <fieldset>
      <legend>Sort Fields</legend>
      <input type="text" name="sort" value="" maxlength="80" size="40"/>
    </fieldset>
    <fieldset>
      <legend>Output Type</legend>
      <select name="writertype">
        <option value="html" selected="true">HTML</option>
        <option value="json">JSON</option>
      </select>
    </fieldset>
    <fieldset>
      <legend>Fields To Return</legend>
      <input type="text" name="fieldlist" value="" maxlength="80" size="40"/>
    </fieldset>
    <fieldset>
      <legend>Fields to Highlight</legend>
      <input type="text" name="highlightfields" value="" maxlength="80" size="40"/>
    </fieldset>
    <fieldset>
      <legend>Fields to Facet</legend>
      <input type="text" name="facetfields" value="" maxlength="80" size="40"/>
    </fieldset>
    <input type="submit" value="Search"/>
  }
}

The resulting input form looks like this:


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
@** app/views/search.scala.html **@
@(result: SearchResult)

@import helper._

@main("Search with ElasticSearch - HTML results") {
  <h2>Search Results</h2>
  <p><b>@result.meta("start") to @result.meta("end") results of @result.meta("numFound") in @result.meta("QTime") ms</b></p>
  <hr/>
  <p><b>JSON Query: </b>@result.meta("query_json")</p>
  <hr/>
  @for(doc <- result.docs) {
    <fieldset>
      <table cellspacing="0" cellpadding="0" border="1" width="100%">
      @for((fieldname, fieldvalue) <- doc) {
        <tr valign="top">
          <td width="20%"><b>@fieldname</b></td>
          <td width="80%">@fieldvalue</td>
        </tr>
      }
      </table>
    </fieldset>
  }
  <hr/>
}

Finally, we come to the part of the application that is not autogenerated by Play2 and which contains all the business logic of the application - the model. Here is the code.

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
// app/models/Searcher.scala
package models

import scala.Array.canBuildFrom

import play.api.libs.json.{Json, JsValue}
import play.api.libs.ws.WS

case class SearchResult(
  meta: Map[String,Any], 
  docs: Seq[Seq[(String,JsValue)]],
  raw: String
)

case class SearchParams(
  index: String,
  query: String,
  filter: String,
  start: Int,
  rows: Int,
  sort: String,
  writertype: String, 
  fieldlist: String,
  highlightfields: String,
  facetfields: String
)

object Searcher {
  
  def search(server: String, params: SearchParams): SearchResult = {
    val payload = Searcher.buildQuery(params)
    val rawResponse = WS.url(server + params.index + 
      "/_search?pretty=true").post(payload).value.get.body
    println("response=" + rawResponse)
    val rsp = Json.parse(rawResponse)
    val meta = (rsp \ "error").asOpt[String] match {
      case Some(x) => Map(
        "error" -> x,
        "status" -> (rsp \ "status").asOpt[Int].get
      )
      case None => Map(
        "QTime" -> (rsp \ "took").asOpt[Int].get,
        "start" -> params.start,
        "end" -> (params.start + params.rows),
        "query_json" -> payload,
        "numFound" -> (rsp \ "hits" \ "total").asOpt[Int].get,
        "maxScore" -> (rsp \ "hits" \ "max_score").asOpt[Float].get
      )
    }
    val docs = if (meta.contains("error")) Seq()
    else {
      val hits = (rsp \ "hits" \ "hits").asOpt[List[JsValue]].get
      val idscores = hits.map(hit => Map(
        "_id" -> (hit \ "_id"),
        "_score" -> (hit \ "_score")))
      val fields = hits.map(hit => 
        (hit \ "_source").asOpt[Map[String,JsValue]].get)
      idscores.zip(fields).
        map(tuple => tuple._1 ++ tuple._2).
        map(doc => doc.toSeq.sortWith((doc1, doc2) => doc1._1 < doc2._1))
    }
    new SearchResult(meta, docs, rawResponse)
  }
  
  def buildQuery(params: SearchParams): String = {
    val queryQuery = Json.toJson(
      if (params.query.isEmpty || "*:*".equals(params.query))
        Map("match_all" -> Map.empty[String,String])
      else Map("query_string" -> Map("query" -> params.query)))
    val queryFilter = if (params.filter.isEmpty) null
      else Json.toJson(Map("query_string" -> Json.toJson(params.filter)))
    val queryFacets = if (params.facetfields.isEmpty) null
      else {
        val fields = params.facetfields.split(",").map(_.trim)
        Json.toJson(fields.zip(fields.
          map(field => Map("terms" -> Map("field" -> field)))).toMap)
      }
    val querySort = if (params.sort.isEmpty) null
      else Json.toJson(params.sort.split(",").map(_.trim).map(field => 
        if (field.toLowerCase.endsWith(" asc") || 
            field.toLowerCase.endsWith(" desc")) 
          (field.split(" ")(0), field.split(" ")(1)) 
        else (field, "")).map(tuple => 
          if (tuple._2.isEmpty) Json.toJson(tuple._1)
          else Json.toJson(Map(tuple._1 -> tuple._2))))  
    val queryFields = if (params.fieldlist.isEmpty) null
      else Json.toJson(params.fieldlist.split(",").map(_.trim))
    val queryHighlight = if (params.highlightfields.isEmpty) null
      else {
        val fields = params.highlightfields.split(",").map(_.trim)
        Json.toJson(Map("fields" -> fields.zip(fields.
          map(field => Map.empty[String,String])).toMap))
      }
    Json.stringify(Json.toJson(Map(
      "from" -> Json.toJson(params.start),
      "size" -> Json.toJson(params.rows),
      "query" -> queryQuery,
      "filter" -> queryFilter,
      "facets" -> queryFacets,
      "sort" -> querySort,
      "fields" -> queryFields,
      "highlight" -> queryHighlight).
      filter(tuple => tuple._2 != null)))
  }
}

The first two are simple case classes, SearchParams and SearchResults are an FBO (Form Backing Object) and DTO (Data Transfer Object) respectively from the Spring world. The search() method takes the ES server URL and the filled in SearchParams object, calls buildQuery() to build the ES Query JSON, then hits the ES server. It then parses the JSON response from ES to create the SearchResult bean, which is passes back to the search Action. The SearchResults object contains a Map containing response metadata, a List of List of key-value pairs which contain the documents, and the raw JSON response from ES.

Here are some screenshots of the results for "hedge fund" from our Enron index that we built using the code from the previous post.






The one on the left shows HTML results (and also shows the JSON query that one would need to use to get the results. The one on the right shows the raw JSON results from the ES server.

Thats all I have for this week. Hope you found it interesting.

Update 2011-11-20 - There were some minor bugs caused by the fields parameter being blank. If the fields parameter is blank, the _source JSON field is returned by ES instead of an array of field objects. The fix is to pass in a "*" (all fields) as the default for the fields parameter. The updated code can be found on my GitHub page.