If you’re already familiar with Hadoop, you can safely skip this section. If you haven’t had the pleasure, Hadoop (http://hadoop.apache.org) is a set of open source projects that deal with large amounts of data in a distributed way. Its Hadoop distributed filesystem (HDFS) and MapReduce subprojects are open source implementations of Google’s GFS and MapReduce.

Google found that several internal groups had been implementing similar functionality in order to solve problems in a distributed way. They saw that it was common to have two phases of operations over distributed data: a map phase and a reduce phase. A map function operates over raw data and produces intermediate values. A reduce function distills those intermediate values in some way, producing the final output for that MapReduce computation. By standardizing on a common framework, they could build more solutions to problems rather than new models of the same wheel.

Doug Cutting decided to write open source implementations of the Google File System (http://labs.google.com/papers/gfs.html) and MapReduce (http://labs.google.com/papers/mapreduce.html), and thus, Hadoop was born. Since then, it has blossomed into myriad tools, all dealing with solutions for big data problems. Today, Hadoop is widely used, by Yahoo!, Facebook, LinkedIn, Twitter, IBM, Rackspace, and many other companies. There is a vibrant community and a growing ecosystem.

Cassandra has built-in support for the Hadoop implementation of MapReduce (http://hadoop.apache.org/mapreduce).

This section covers details on how to write a simple MapReduce job over data stored in Cassandra using the Java language. We also briefly cover how to output data into Cassandra and discuss ongoing progress with using Cassandra with Hadoop Streaming for languages beyond Java.

For convenience, the word count MapReduce example can be run locally against a single Cassandra node. However, for more information on how to configure Cassandra and Hadoop to run MapReduce in a more distributed fashion, see the section Cluster Configuration.

Word count is one of the examples given in the MapReduce paper and is the starting point for many who are new to the framework. It takes a body of text and counts the occurrences of each distinct word. Here we provide some code to perform a word count over data contained in Cassandra. A working example of word count is also included in the Cassandra source download.

First we need a Mapper class, shown in Example 12-1.

public static class TokenizerMapper extends Mapper<byte[], 
    SortedMap<byte[], IColumn>, Text, IntWritable> {

  private final static IntWritable one = new IntWritable(1);
  private Text word = new Text();
  private String columnName;

  public void map(byte[] key, SortedMap<byte[], IColumn> columns, Context context)
    throws IOException, InterruptedException {

    IColumn column = columns.get(columnName.getBytes());
    String value = new String(column.value());
    StringTokenizer itr = new StringTokenizer(value);

    while (itr.hasMoreTokens()) {
      word.set(itr.nextToken());
      context.write(word, one);
    }
  }

  protected void setup(Context context)
    throws IOException, InterruptedException {

    this.columnName = context.getConfiguration().get(“column_name”);
 }
}

Readers familiar with MapReduce programs will notice how familiar this mapper looks. In this case, the inputs to the mapper are row keys and associated row values from Cassandra. Row values in the world of Cassandra are simply maps containing the column information. In addition to the word count code itself, we override the setup method to set the column name we are looking for. The rest of the mapper code is generic to any word count implementation.

Next, let’s look at a Reducer implementation for our word count, shown in Example 12-2.

public static class IntSumReducer extends 
  Reducer<Text, IntWritable, Text, IntWritable> {

  private IntWritable result = new IntWritable();

  public void reduce(Text key, Iterable<IntWritable> values, Context context)
    throws IOException, InterruptedException {

    int sum = 0;
    for (IntWritable val : values) {
      sum += val.get();
    }
    result.set(sum);
    context.write(key, result);
  }
}

There should be nothing surprising in this reducer; nothing is Cassandra-specific.

Finally, we get to the class that runs our MapReduce program, shown in Example 12-3.

public class WordCount extends Configured implements Tool {

  public int run(String[] args) throws Exception {
    Job job = new Job(getConf(), “wordcount”);
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    job.setInputFormatClass(ColumnFamilyInputFormat.class);

    FileOutputFormat.setOutputPath(job, new Path(“/tmp/word_count”));
    ConfigHelper.setThriftContact(job.getConfiguration(), “localhost”, 9160);
    ConfigHelper.setInputColumnFamily(
        job.getConfiguration(), “Keyspace1”, “Standard1”);
    
    SlicePredicate predicate = new SlicePredicate().setColumn_names(
        Arrays.asList(columnName.getBytes()));

    ConfigHelper.setInputSlicePredicate(job.getConfiguration(), predicate);
    job.waitForCompletion(true);

    return 0;
  }
}

There are a few things to note about the WordCount class that go beyond a boilerplate word count. We need to set the InputFormat to Cassandra’s ColumnFamilyInputFormat, along with the column name that our mapper is looking for. Cassandra includes something called a ConfigHelper that provides a way to set properties we’ll need, such as the Thrift contact information (server and port), the keyspace, and the column family. It also allows us to set our slice predicate.

MapReduce is a great abstraction for developers so that they can worry less about the details of distributed computing and more about the problems they are trying to solve. Over time, an even more abstracted toolset has emerged. Pig and Hive operate at a level above MapReduce and allow developers to perform more complex analytics more easily. Both of these frameworks can operate against data in Cassandra.

LOAD 'cassandra://Keyspace1/Standard1' USING CassandraStorage() 
as (key:chararray, cols:bag{col:tuple(name:bytearray, value:bytearray)});
cols = FOREACH rows GENERATE flatten(cols) as (name, value);
words = FOREACH cols GENERATE flatten(TOKENIZE((chararray) value)) as word;
grouped = GROUP words BY word;
counts = FOREACH grouped GENERATE group, COUNT(words) as count;
ordered = ORDER counts BY count DESC;
topten = LIMIT ordered 10;
dump topten;

MapReduce and other tools can run in a nondistributed way for trying things out or troubleshooting a problem. However, in order to run in a production environment, you’ll want to install Hadoop in your Cassandra cluster as well. Although a comprehensive discussion of Hadoop installation and configuration is outside the scope of this chapter, we do go over how to configure Cassandra alongside Hadoop for best performance. Readers can find more about Hadoop configuration at http://hadoop.apache.org or in Tom White’s excellent reference, Hadoop: The Definitive Guide (O’Reilly).

Because Hadoop has some unfamiliar terminology, here are some useful definitions:

Like Cassandra, Hadoop is a distributed system. The MapReduce jobtracker spreads tasks across the cluster, preferably near the data it needs. When a jobtracker initiates tasks, it looks to HDFS to provide it with information about where that data is stored. Similarly, Cassandra’s built-in Hadoop integration provides the jobtracker with data locality information so that tasks can be close to the data.

In order to achieve this data locality, Cassandra nodes must also be part of a Hadoop cluster. The namenode and jobtracker can reside on a server outside your Cassandra cluster. Cassandra nodes will need to be part of the cluster by running a tasktracker process on each node. Then, when a MapReduce job is initiated, the jobtracker can query Cassandra for locality of the data when it splits up the map and reduce tasks.

A four-node Cassandra cluster with tasktracker processes running on each Cassandra node is shown in Figure 12-1. At least one node in the cluster needs to be running the datanode process. There is a light dependency on HDFS for small amounts of data (the distributed cache), and a single datanode should suffice. External to the cluster is the server running the Hadoop namenode and jobtracker.

Figure 12-1. Four-node Cassandra cluster with an external namenode/jobtracker

When a job is initiated from a client machine, it goes to the jobtracker. The jobtracker receives information about the data source from when the job is submitted, via the configuration options mentioned earlier. At that point, it can use Cassandra’s ColumnFamilyRecordReader and ColumnFamilySplit to get the physical locations of different segments of data in the cluster. Then, it can use that location to split up the tasks among the nodes in the cluster, preferring to run tasks on nodes where the associated data resides.

Finally, when creating jobs for MapReduce to execute (either directly or via Pig), the Hadoop configuration needs to point to the namenode/jobtracker (in the Hadoop configuration files) and the Cassandra configuration options. The cluster will be able to handle the integration from there.

To help you understand why Cassandra/Hadoop integration is interesting and useful, here we include a couple of use cases from out in the Cassandra community.

In this chapter, we examined Hadoop, the open source implementation of the Google MapReduce algorithm. We took a brief look at Hadoop basics and how to integrate it with Cassandra by running the word count example that ships with Cassandra, and also saw how to use Pig, a simple and concise language to express MapReduce jobs.

Finally, we got a picture of how a few different companies are using Hadoop and Cassandra together.