Ikai Lan says

I say things!

Archive for July 2010

Issuing App Engine datastore queries with the Low-Level API

Last time, I wrote an introduction to using the low-level API for creating entities, setting keys, and getting keys by value.

Basic queries and sorts

These are useful when we know the keys, but its often very useful to be able to query entities by their properties. Consider the Person entities we created for the last example, Alice and Bob:

Entity alice = new Entity("Alice", "Person");
 alice.setProperty("gender", "female");
 alice.setProperty("age", 20);

 Entity bob = new Entity(“Person”, “Bob”);
 bob.setProperty("gender", "male");
 bob.setProperty("age", "23");

 DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
 datastore.put(alice);
 datastore.put(bob);

Let’s create a query to find the first 10 Persons that are female and sort them by age ascending. How would we write this?

Query findFemalesQuery = new Query("Person");
findFemalesQuery.addFilter("gender", FilterOperator.EQUAL, "female");
findFemalesQuery.addSort("age", SortDirection.ASCENDING);
datastore.prepare(findFemalesQuery).asList(FetchOptions.Builder.withLimit(10));

Here are the steps we took:

  1. Created a Query object, specifying the Query kind
  2. Added a QueryFilter. Note that this is typesafe. We specify the enum representing the FilterOperator we want to use
  3. Added a QuerySort. Again, like the QueryFilter, we select the property to sort on as well as an enum representing either an ascending order or descending order.
  4. We prepare the query. On this result we return it as either an Iterator or as a List of Entities. On this method we can either execute the default query, or we can pass a set of options. In the example above, we use FetchOptions.Builder to set the only option we care about: the limit. We only want 10, so we call withLimit() and pass it 10.

The query interface works well because it’s typesafe where the datastore is typesafe, and not so when the datastore is not – you won’t get errors at runtime because you misspelled “WHERE”, for instance, but you have to be careful not to misspell the properties you are looking for. The flexibility of this interface means that no longer are we constrained by the “every object must have the same bag of properties” frame of thinking. Furthermore, because we don’t need to know the property names apriori (we can use getProperties() and return a Map), we can iterate through this and figure out the keys/value pairs at runtime. This leads to some very powerful abstractions.

Doing a keys only query

It sometimes makes sense for us to only retrieve the keys in a given query. It’s actually incredibly easy, so as long as we know what to expect:

Query findFemalesQuery = new Query("Person");
findFemalesQuery.addFilter("gender", FilterOperator.EQUAL, "female");
findFemalesQuery.addSort("age", SortDirection.ASCENDING);
findFemalesQuery.setKeysOnly();

List<Entity> results = datastore.prepare(findFemalesQuery).asList(
FetchOptions.Builder.withLimit(10));

The only code that’s different in creating the Query object is that we call setKeysOnly(). This still returns a List of entity objects with only the Kind and Key populated. If we wrote a test for this, it would look like this:

Entity alice = results.get(0);
assertEquals("Return Key for Entity", KeyFactory.createKey("Person", "Alice"), alice.getKey());
assertNull("Should not return female property", alice.getProperty("gender"));
assertEquals("Returns Entities with no properties", 0, alice.getProperties().size());

Only the Kind and Key are populated in these Entity objects. Even though the API looks similar, under the hood, the behavior is completely different. Recall how queries work underneath the hood:

  1. Traverse an index and retrieve keys
  2. Using those keys, fetch the entities from the datastore

The time to do a query depends on the index traversal time as well as the number of entities to retrieve. In a keys only query, this is what happens:

  1. Traverse an index and retrieve keys

We completely eliminate step 2 from the process. If all we want is Key information or are counting entities (and the count can be done using only indexes), this is the approach we would take.

Ancestor Queries

Let’s pretend Alice and Bob have child entities:

Entity madHatter = new Entity("Friend", "Mad Hatter", alice.getKey());
Entity doormouse = new Entity("Friend", "Doormouse", alice.getKey());
Entity chesireCat = new Entity("Friend", "Chesire Cat", alice.getKey());

Entity redQueen = new Entity("Friend", "Red Queen", bob.getKey());

datastore.put(madHatter);
datastore.put(doormouse);
datastore.put(chesireCat);
datastore.put(redQueen);

Alice now has Friends Mad Hatter, Doormouse and the Chesire Cat as child entities, while Bob has on the Red Queen. How do we find all friends of Alice or Bob? Like so:

Query friendsOfAliceQuery = new Query("Friend");
friendsOfAliceQuery.setAncestor(alice.getKey());

List<Entity> results = datastore.prepare(friendsOfAliceQuery).asList(FetchOptions.Builder.withDefaults());

Query friendsOfBobQuery = new Query("Friend");
friendsOfBobQuery.setAncestor(bob.getKey());

results = datastore.prepare(friendsOfBobQuery).asList(FetchOptions.Builder.withDefaults());

What’s great about these queries is that the datastore knows exactly where to start. Because keys embed parent Key information – Mad Hatter, Doormouse and the Chesire Cat all have “Alice” as a prefix in their key (this is also why you cannot change an entity’s entity group after creation), we know that we just need to start the query from Alice’s Key and just traverse entities with a Key greater than Alice. It’s also a great way of organizing data. Just be aware that too many transactions on a single entity group will destroy your throughput, so design for as small entity groups as possible.

Summary

Hopefully this blog post explains a few more features of the low-level API. Understanding the low-level API is an important step in understanding the datastore, and understanding the datastore is a critical step for learning how to build efficient, optimized applications for App Engine.

Written by Ikai Lan

July 13, 2010 at 4:43 pm

Google App Engine Tips and Tricks: Prebuilding Indexes using a non default version

(This’ll be a shorter post than usual.)

Waiting for indexes to build can be drag; indexes need to be built before Entities even exist and can take longer than needed if the global index building workflow is backed up since mass building is a shared resource.

One little known trick is to pre-build indexes before your application needs them by deploying a non-default version. Your application can have many versions. In Java App Engine, this is defined in the version tag of appengine-web.xml. In Python, this is defined in the version YAML element. The Java Eclipse plugin even has a screen where the version can be set (Click the App Engine icon, then “Project Settings”:

Because all applications share the same datastore, the required indexes will be built once your push your application containing the indexes configuration file with the new, required indexes. Hopefully, by the time you are ready to push your real version, the indexes will have completed building.

In general, it is a best practice to maintain a staging version of your application for testing against live data. App Engine makes this so easy it’s trivial: deploy code tagged with a new “version”. Your application is accessible at http://VERSION.latest.APPID.appspot.com (note that VERSION is a String, not a integer or decimal number) – this is a handy and powerful trick to validating a new test or staging version. When you have enough confidence in your application, browse to the Admin Console, click the radio button associated with your new version, and click “Make Default.”

Versioning has never been so easy. No configuring load balancers, rolling deploys, symlinking, restarting edge caches, etc.

Happy hacking.

Written by Ikai Lan

July 12, 2010 at 12:34 pm

Using the Java Mapper Framework for App Engine

The recently released Mapper framework is the first part of App Engine’s mapreduce offering. In this post, we’ll be discussing some of the types of operations we can perform using this framework and how easily they can be done.

Introduction to Map Reduce

If you aren’t familiar with Map Reduce, read more about it from a high level from Wikipedia here. The official paper can be downloaded from this site if you’re interested in a more technical discussion.

The simplest breakdown of MapReduce is as follows:

  1. Take a large dataset and break it into pieces, mapping individual pieces of data
  2. Work on those mapped datasets and reduce them into the form you need

A simple example here is full text indexing. Suppose we wanted create indexes from existing text documents. We would use the Map step to iterate over every document and “map” each phrase or term to a document, then we would “reduce” the mappings by writing them to an index. Map/reduce problems have the advantage of not only being easy to conceptualize as problems that can be distributed and parallelized, but also because there are frameworks that support many of the administrative functions of map-reduce: failure recovery, distribution of work, tracking status of jobs, reporting and so forth. The appengine-mapreduce project seeks to provide as many of these features as possible while making it as easy as possible for developers to write large batch processing jobs without having to think about the plumbing details.

But I only have Map available!

Yes, this is true  - as of the writing of this post, only the “map” step exists, hence why it’s currently referred to as the “Mapper API”. That doesn’t mean it’s not useful. For starters, it is a very easy way to perform some operation on every single Entity of a given Kind in your datastore in parallel. What would you have to build for yourself if Mapper weren’t available?

  1. Begin querying over every Entity in chained Task Queues
  2. Store beginning and end cursors (introduced in 1.3.5)
  3. Create tasks to work with chunks of your datastore
  4. Write the code to manipulate your data
  5. Build an interface to control your batch jobs
  6. Build a callback system for your multitudes of parallelized workers to call when the entire task has completed

It’s certainly not a trivial amount of work. Some things you can do very easily with the Mapper library include:

  • Modify some property or set of properties for every Entity of a given Kind
  • Delete all entities of a single Kind – the functional equivalent of a “DROP TABLE” if you were using a relational database
  • Count the occurrences of some property across every single Entity of a given Kind in your datastore

We’ll go through a few of these examples in this post.

Our Sample application

Our sample application will be a modified version of the Guestbook demo. We’ll add a few additional properties. For simplicity, we’ll use the low-level API, since the Mapper API also uses the low-level API. You can see this application here:
The code is also available to clone via Github if you’d like to follow along.

How to define a Mapper

There are three steps to defining a Mapper:

  1. Download, build and place the appengine-mapreduce JAR files in your WEB-INF/lib directory and add them to your build path. You only need to do this once per project. The steps for doing this are on the “Getting Started” page for Java. You’ll need all the JAR files that are built.
  2. Make sure that we have a DESCENDING index created on Key. This is important! If we run our Mapper locally, this’ll automatically be created in our datastore-indexes.xml file when we deploy our application. One trick to ensure that indexes get built before they are needed, at least in a live application, is to create and deploy an application with the new index configuration to a non-default version. Because all versions use the same datastore and the same set of indexes, this will schedule the index to be built before we need it in the live version. When it has completed, we simply switch the default version over, and we’re ready to roll.
  3. Create your Mapper class
  4. Configure your Mapper class in mapreduce.xml

We’ll go over steps 3 and 4 in each example.

Example 1: Changing a property on every Entity (Naive way)

(You can even use this technique if you just need to change a property on a large set of Entities).

Assuming you’ve already set up your environment for the Mapper servlet, you can dive right in. Let’s create a Mapper classes that goes through every Entity of a given Kind and converts the “comment” property to use all lowercase letters. We’ll also add a timestamp for when we modified this Entity. In this first example, we’ll do this the naive way. This is a very good way to introduce you to very simple mutations on all your Entities using Mapper.

Note that this requires some familiarity with the Low-Level API. Don’t worry – entities edited or saved using the low-level API are accessible via managed persistence interface such as JDO/JPA (and vice versa). If you aren’t familiar with the low-level API, you can read more about it here on the Javadocs.

The first thing we’ll have to do is define a Mapper. We tried as much as possible to mimic Hadoop’s Mapper class. We’ll be subclassing AppEngineMapper, which is itself a subclass of Hadoop’s Mapper. The meat of this class is the map() method, which we’ll be overriding. We’ll also override the taskSetup() lifecycle callback. We’ll be using this to initialize our DatastoreService, though we could probably initialize it in the body of the map() method itself. The other methods are taskCleanup(), setup() and cleanup() – examples here. Let’s have a look at our code below:

package com.ikai.mapperdemo.mappers;

import java.util.Date;
import java.util.logging.Logger;

import org.apache.hadoop.io.NullWritable;

import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreServiceFactory;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.tools.mapreduce.AppEngineMapper;

/**
 *
 * This mapper changes all Strings to lowercase Strings, sets
 * a timestamp, and reputs them into the Datastore. The reason
 * this is a "Naive" Mapper is because it doesn't make use of
 * Mutation Pools, which can do these operations in batch instead
 * of individually.
 *
 * @author Ikai Lan
 *
 */
public class NaiveToLowercaseMapper extends
		AppEngineMapper<Key, Entity, NullWritable, NullWritable> {
	private static final Logger log = Logger
			.getLogger(NaiveToLowercaseMapper.class.getName());

	private DatastoreService datastore;

	@Override
	public void taskSetup(Context context) {
		this.datastore = DatastoreServiceFactory.getDatastoreService();
	}

	@Override
	public void map(Key key, Entity value, Context context) {
		log.info("Mapping key: " + key);

		if (value.hasProperty("comment")) {
			String comment = (String) value.getProperty("comment");
			comment = comment.toLowerCase();
			value.setProperty("comment", comment);
			value.setProperty("updatedAt", new Date());

			datastore.put(value);

		}
	}
}

Notice that this map method takes 3 parameters:

Key key – this is the datastore Key for the Entity we are about to perform an operation on. Mostly this exists for API compatibility with Hadoop, but we don’t really need it yet. For iterating over datastore Entities, we don’t really need this, because we *could* use this to look up the Entity, but we don’t have to because …

Entity value – … because we actually get the Entity already. If we did a lookup for the Entity, we’d double the amount of lookups we do per Entity. We can certainly use the Key to do a lookup using a PersistenceManager or EntityManager and have a populated, typesafe Entity object, but from an efficiency standpoint we’d be doubling our work for some JDO/JPA sugar.

Context context – We don’t need this in our example, but it’s easy to think of the Context as giving us access to “global” values such as temporary variables and configuration files. For a later example in this post, we’ll be using the Context to store a global value in a counter and increment it. For this example, it’s unused.

If you’re familiar at all with the low-level API, this will look very straightfoward (again, I highly encourage you to read the docs). We take an entity, add 2 properties to it, then re-put() the Entity back into the datastore.

Now let’s add this job to mapreduce.xml:

<configurations>
  <configuration name="Naive Mass toLowercase()">
    <property>
      <name>mapreduce.map.class</name>

      <!--  Set this to be your Mapper class  -->
      <value>com.ikai.mapperdemo.mappers.NaiveToLowercaseMapper</value>
    </property>

    <!--  This is a default tool that lets us iterate over datastore entities -->
    <property>
      <name>mapreduce.inputformat.class</name>
      <value>com.google.appengine.tools.mapreduce.DatastoreInputFormat</value>
    </property>

    <property>
      <name human="Entity Kind to Map Over">mapreduce.mapper.inputformat.datastoreinputformat.entitykind</name>
      <value template="optional">Comment</value>
    </property>
  </configuration>
</configurations>

It looks complex, but it’s really not. We define a configuration element and name the job. The name of the job is also the name we’ll see in the GUI when we fire off the job. We need 3 sets of property elements under this element, which are just name/value pairs. Let’s go over each one we used:

Name: mapreduce.map.class
Value: com.ikai.mapperdemo.mappers.NaiveToLowercaseMapper
This one is straightforward – we provide the name of an AppEngineMapper subclass with the map() method we want run.

Name: mapreduce.inputformat.class
Value: com.google.appengine.tools.mapreduce.DatastoreInputFormat
This is a class that takes some input to map over. DatastoreInputFormat is provided by appengine-mapreduce, but it is possible for us to define our own input formatter. For guidance, check out the source of DatastoreInputFormat here.

In a more advanced example (ahem, future blog post), we’ll discuss building our own InputFormat to read from another source such as the Blobstore. For our examples in this post, we won’t need anything beyond DatastoreInputFormat.

Name: mapreduce.mapper.inputformat.datastoreinputformat.entitykind
Value: Comment
This input is specific to DatastoreInputFormat. It tells DatastoreInputFormat which Entity Kind to iterate over. Note that in the mapper console, a user can type in the name of a Kind or edit this Field to reflect the value they want. We can’t leave this blank, though, if we want this to work.

If we browse to the URI at which we’ve defined the Mapper console (in our case /mapper), we see something that looks like this:

“Running jobs” appears when we click “Run”. We can click “Detail” to see the progress of our job, or we can “Abort” to quit the job. Note that aborting a job won’t revert our Entities! We’ll end up with a partially run job if we run a giant mutation, so we’ll have to be cognizant of this when we use this tool.

When the job completes, we’ll take a look at our Comments. Sure enough, they are now all lowercase.

Example 2: Changing a property on every Entity using Mutation Pools

There’s a reason the Mapper in Example 1 is called a Naive Mapper: because it doesn’t take advantage of mutation pools. As we all know, App Engine’s datastore is capable of handling operations in parallel using batched calls. We’re already doing work in parallel by specifying shards, but we’ll want to use batched calls when possible. We do this by adding the mutations we want to a mutation pool, then, periodically as the pool hits a certain size, we flush all the writes to the datastore with a single call instead of individually. This has the advantage of making our map() call as fast as possible, since all we’re really doing is making a list of operations to perform all at once when the system is good and ready. Let’s define the XML file first assuming we call the class PooledToLowercaseMapper:

  <configuration name="Mass toLowercase() with Mutation Pool">
    <property>
      <name>mapreduce.map.class</name>

      <!--  Set this to be your Mapper class  -->
      <value>com.ikai.mapperdemo.mappers.PooledToLowercaseMapper</value>
    </property>

    <!--  This is a default tool that lets us iterate over datastore entities -->
    <property>
      <name>mapreduce.inputformat.class</name>
      <value>com.google.appengine.tools.mapreduce.DatastoreInputFormat</value>
    </property>

    <property>
      <name human="Entity Kind to Map Over">mapreduce.mapper.inputformat.datastoreinputformat.entitykind</name>
      <value template="optional">Comment</value>
    </property>

  </configuration>

It looks almost exactly the same. That’s because the meat is in what we do in the actually class itself:

package com.ikai.mapperdemo.mappers;

import java.util.Date;
import java.util.logging.Logger;

import org.apache.hadoop.io.NullWritable;

import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.tools.mapreduce.AppEngineMapper;
import com.google.appengine.tools.mapreduce.DatastoreMutationPool;

/**
 *
 * The functionality of this is exactly the same as in {@link NaiveToLowercaseMapper}.
 * The advantage here is that since a {@link DatastoreMutationPool} is used, mutations
 * can be done in batch, saving API calls.
 *
 * @author Ikai Lan
 *
 */
public class PooledToLowercaseMapper extends
		AppEngineMapper<Key, Entity, NullWritable, NullWritable> {
	private static final Logger log = Logger
			.getLogger(PooledToLowercaseMapper.class.getName());

	@Override
	public void map(Key key, Entity value, Context context) {
		log.info("Mapping key: " + key);

		if (value.hasProperty("comment")) {
			String comment = (String) value.getProperty("comment");
			comment = comment.toLowerCase();
			value.setProperty("comment", comment);
			value.setProperty("updatedAt", new Date());

			DatastoreMutationPool mutationPool = this.getAppEngineContext(
					context).getMutationPool();
			mutationPool.put(value);
		}
	}
}

Everything looks example the same until these lines:

DatastoreMutationPool mutationPool = this.getAppEngineContext(context).getMutationPool();
mutationPool.put(value);

Aha! So we finally put the context to use. Granted, we use the context as a parameter to another, more useful method, but at least we’re using it.  We acquire a DatastoreMutationPool using the getAppEngineContext(context).getMutationPool() method, then we just call put() and pass the changed entity. DatastoreMutationPool is defined here and is open source.

The interface is similar to that of DatastoreService. There’s not a lot of fancy stuff going on here. put(), as we’ve seen, is defined. get() isn’t, because, well, that method makes no sense in this context. delete() is defined, which brings me to my bonus section:

Bonus Example 2: Delete all Entities of a given Kind

One of the most common questions asked in the group is, “How do I drop table?” Usually, this question is asked by new App Engine developers who don’t yet understand that the datastore is a distributed key-value store and not a relational database. But it’s also a legitimate use case. What if you just wanted to nuke all Entities of a given Kind? Prior to Mapper, you would have had to write your own handler to take care of this. Mapper makes this very easy. Here’s what a generic “DeleteAllMapper” would look like. This will work with *any* Entity Kind:

package com.ikai.mapperdemo.mappers;

import java.util.logging.Logger;

import org.apache.hadoop.io.NullWritable;

import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.tools.mapreduce.AppEngineMapper;
import com.google.appengine.tools.mapreduce.DatastoreMutationPool;

/**
 *
 * This Mapper deletes all Entities of a given kind. It simulates the
 * DROP TABLE functionality asked for by developers.
 *
 * @author Ikai Lan
 *
 */
public class DeleteAllMapper extends
		AppEngineMapper<Key, Entity, NullWritable, NullWritable> {
	private static final Logger log = Logger.getLogger(DeleteAllMapper.class
			.getName());

	@Override
	public void map(Key key, Entity value, Context context) {
		log.info("Adding key to deletion pool: " + key);
		DatastoreMutationPool mutationPool = this.getAppEngineContext(context)
				.getMutationPool();
		mutationPool.delete(value.getKey());
	}
}

That’s it! We wire it up the same way we wire up other Mappers:


  <configuration name="Delete all Entities">
    <property>
      <name>mapreduce.map.class</name>

      <!--  Set this to be your Mapper class  -->
      <value>com.ikai.mapperdemo.mappers.DeleteAllMapper</value>
    </property>

    <!--  This is a default tool that lets us iterate over datastore entities -->
    <property>
      <name>mapreduce.inputformat.class</name>
      <value>com.google.appengine.tools.mapreduce.DatastoreInputFormat</value>
    </property>

    <property>
      <name human="Entity Kind to Map Over">mapreduce.mapper.inputformat.datastoreinputformat.entitykind</name>
      <value template="optional">Comment</value>
    </property>
  </configuration>

I’ve separated each out into its own mapreduce.xml, but this isn’t necessary. A given App Engine project can have multiple configuration elements defined. That’s why there’s a dropdown list in the Mapreduce console GUI.

Example 3: Taking more user input in the Mapper console and counting

Our next example covers using counters in the context. Let’s say that we wanted to allow the User to enter a String, then we iterate over every Entity searching for occurrences of that Substring on-the-fly and not with pre-built indexes. First, let’s discuss the XML configuration we use:

  <configuration name="Count words in all Comments">
    <property>
      <name>mapreduce.map.class</name>

      <!--  Set this to be your Mapper class  -->
      <value>com.ikai.mapperdemo.mappers.CountWordsMapper</value>
    </property>

    <property>
    	<!--  This is the URL to call after the entire Mapper has run -->
    	<name>mapreduce.appengine.donecallback.url</name>
    	<value>/callbacks/word_count_completed</value>
    </property>

    <!--  This is a default tool that lets us iterate over datastore entities -->
    <property>
      <name>mapreduce.inputformat.class</name>
      <value>com.google.appengine.tools.mapreduce.DatastoreInputFormat</value>
    </property>

    <property>
      <name human="Entity Kind to Map Over">mapreduce.mapper.inputformat.datastoreinputformat.entitykind</name>
      <value template="optional">Comment</value>
    </property>

  </configuration>

There’s one new name/value pair:
Name: mapreduce.mapper.counter.substringtarget
Value: Substring
We can pick any name or value we want. We just pick this one because it makes sense. We’ll retrieve this value in the Mapper via the Context. This causes an extra text field to appear in the Mapper console:


The Mapper is below:

package com.ikai.mapperdemo.mappers;

import java.util.logging.Logger;

import org.apache.hadoop.io.NullWritable;

import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.tools.mapreduce.AppEngineMapper;

/**
 *
 * This Mapper takes some input and counts the number of Comments which
 * contain that substring.
 *
 * @author Ikai Lan
 *
 */
public class SubstringMatcherMapper extends
		AppEngineMapper<Key, Entity, NullWritable, NullWritable> {
	private static final Logger log = Logger.getLogger(SubstringMatcherMapper.class
			.getName());

	/*
	 * Get the target that we want to match on and count the number of Comments that
	 * match it
	 */
	@Override
	public void map(Key key, Entity value, Context context) {

	    String substringToMatch = context.getConfiguration().get("mapreduce.mapper.counter.substringtarget");

		String comment = (String) value.getProperty("comment");
		if (comment != null) {
			if(comment.contains(substringToMatch)) {
				log.info("Found match in: " + comment);
				context.getCounter("SubstringMatch", "count").increment(1);
			}
		}

	}
}

We retrieve the value entered by the user with this line of code:

context.getConfiguration().get("mapreduce.mapper.counter.substringtarget");

If the comment we’re current working on contains the substring, we want to increment our count. The context object has a getCounter() method that returns a counter we can increment or decrement:

context.getCounter("SubstringMatch", "count").increment(1);

When our job completes running, we can see the total count when we click “Detail” on the completed job:

More likely than not, however, we’ll want to store this number back in the datastore or do something with it besides stick it into a status page. Good that we mention that …

Example 4: Completion callbacks and JobContexts

Let’s modify Example 3 a bit. Suppose now we want to count the total number of words across all comments. We’ll need to use a counter. But suppose that instead of just displaying it in a console page, we want that number to get stored into the datastore again. Much like Task Queues, incoming email and XMPP, the callback is event driven, and therefore uses the HTTP request to an app URI model to dispatch. That is – we’ll define a servlet with a doPost() handler and read the input out of the parameters.

The first thing we’ll need to do is configure our Mapper to fire off the callback when done. We do this in mapreduce.xml:

  <configuration name="Count substring matches in all Comments">
    <property>
      <name>mapreduce.map.class</name>

      <!--  Set this to be your Mapper class  -->
      <value>com.ikai.mapperdemo.mappers.SubstringMatcherMapper</value>
    </property>

    <!--  This is a default tool that lets us iterate over datastore entities -->
    <property>
      <name>mapreduce.inputformat.class</name>
      <value>com.google.appengine.tools.mapreduce.DatastoreInputFormat</value>
    </property>

    <property>
      <name human="Entity Kind to Map Over">mapreduce.mapper.inputformat.datastoreinputformat.entitykind</name>
      <value template="optional">Comment</value>
    </property>

    <property>
      <name human="Search for substring">mapreduce.mapper.counter.substringtarget</name>
      <value template="optional">Substring</value>
    </property>

  </configuration>

Here’s the property we care about:

Name: mapreduce.appengine.donecallback.url
Value: /callbacks/word_count_completed

The value of this can map to any URI in your application. Just be sure that URI points to the Servlet that will be handling your callback. Let’s define the Mapper class:

package com.ikai.mapperdemo.mappers;

import java.util.logging.Logger;

import org.apache.hadoop.io.NullWritable;

import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.tools.mapreduce.AppEngineMapper;

/**
 *
 * This mapper counts the number of total words across all comments. It cheats a
 * bit by just splitting on whitespace and just using the length. This mapper
 * demonstrates use of counters as well as using a completion callback.
 *
 * @author Ikai Lan
 *
 */
public class CountWordsMapper extends
		AppEngineMapper<Key, Entity, NullWritable, NullWritable> {
	private static final Logger log = Logger.getLogger(CountWordsMapper.class
			.getName());

	/*
	 * This is a bit of a lazy implementation more to prove a point than to
	 * actually be correct. Split on whitespace, count words
	 */
	@Override
	public void map(Key key, Entity value, Context context) {

		String comment = (String) value.getProperty("comment");
		if (comment != null) {
			String[] words = comment.split("\\s+");
			int wordCount = words.length;

			// Takes a "group" and a "counter"
			// We'll use these later to store the final count back in the
			// datastore
			context.getCounter("CommentWords", "count").increment(wordCount);
		}

	}
}

Not a lot that’s new here. We use the context again to store a counter. Note that we can increment by any value, not just 1.

Let’s take a look at what our servlet looks like that handles this callback:

package com.ikai.mapperdemo.servlets;

import java.io.IOException;
import java.util.Date;

import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.CounterGroup;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.JobID;

import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreServiceFactory;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityNotFoundException;
import com.google.appengine.tools.mapreduce.MapReduceState;
import com.ikai.mapperdemo.mappers.CountWordsMapper;

/**
 * This is the servlet that takes care of any processing we have to do after we
 * have finished running {@link CountWordsMapper}.
 *
 * This is just a standard servlet - we can do anything we want here. We can use
 * any App Engine API such as email or XMPP, for instance, to notify an
 * administrator. We could also store a final state into the datastore - in
 * fact, that is what this example below does.
 *
 * @author Ikai Lan
 *
 */
@SuppressWarnings("serial")
public class WordCountCompletedCallbackServlet extends HttpServlet {

	public void doPost(HttpServletRequest req, HttpServletResponse resp)
			throws IOException {

		String jobIdName = req.getParameter("job_id");
		JobID jobId = JobID.forName(jobIdName);

		// A future iteration of this will likely contain a default
		// option if we don't care which DatastoreService instance we use.
		DatastoreService datastore = DatastoreServiceFactory
				.getDatastoreService();
		try {

			// We get the state back from the job_id parameter. The state is
			// serialized and stored in the datastore, so we pass an instance
			// of the datastore service.
			MapReduceState mrState = MapReduceState.getMapReduceStateFromJobID(
					datastore, jobId);

			// There's a bit of ceremony to get the actual counter. This
			// example is intentionally verbose for clarity. First get all the
			// Counters,
			// then we get the CounterGroup, then we get the Counter, then we
			// get the count.
			Counters counters = mrState.getCounters();
			CounterGroup counterGroup = counters.getGroup("CommentWords");
			Counter counter = counterGroup.findCounter("count");
			long wordCount = counter.getValue(); // Finally!

			// Let's create a special datastore Entity for this value so
			// we can reference it on the ViewComments page
			Entity totalCountEntity = new Entity("TotalWordCount",
					"total_word_count");
			totalCountEntity.setProperty("count", wordCount);

			// Now we timestamp this bad boy
			totalCountEntity.setProperty("updatedAt", new Date());
			datastore.put(totalCountEntity);

		} catch (EntityNotFoundException e) {
			throw new IOException("No datastore state");
		}

	}

}

The JobID comes as a String parameter. We get it like so:

String jobIdName = req.getParameter("job_id");
JobID jobId = JobID.forName(jobIdName);

Be aware of the imports used. Your IDE may import the wrong class, as there is a deprecated JobID and a non-deprecated version.

Once you have the JobID, you use it to retrieve the MapReduceState:

MapReduceState mrState = MapReduceState.getMapReduceStateFromJobID(datastore, jobId);

From the MapReduceState object, we have to perform a bit of a ceremony to get what we want. We need to:

1. Fetch the Counters from the MapReduceState
2. Fetch the appropriate CounterGroup from the Counters object
3. Fetch the named Counter from the CounterGroup
4. Fetch the value from the Counter

Here’s what it looks like in code:

Counters counters = mrState.getCounters();
CounterGroup counterGroup = counters.getGroup("CommentWords");
Counter counter = counterGroup.findCounter("count");
long wordCount = counter.getValue();

We can now do what we want with this count. In our servlet example, we save it to a datastore Entity and use it later on.

Get the code

You’re undoubtedly ready to start playing with this thing. You’ve got everything you need to know. First, here’s the getting started page for appengine-mapreduce in Java:

Here’s my sample source code on GitHub.

Summary

So there you have it: an easy to use tool for mapping operations across entire Entity Kinds. There are still a lot of topics to cover, and we’ll likely explore them in a future article. For instance, I didn’t have a chance to cover building your own InputFormat class. We’re still hard at work extending this framework (such as the “Shuffle” and “Reduce” phases), so please post your feedback in the App Engine groups or file bugs in the issue tracker.

Written by Ikai Lan

July 9, 2010 at 3:35 pm

Posted in App Engine, Java, Java