Using Apache Cassandra with Apache Hadoop

I am currently working on a data analytics website for my own educational purposes and to fulfil my hacking/learning needs, I decided to use Apache Cassandra as the input/output storage engine for and Apache Hadoop map/reduce job.

The job in question is as simple as it gets: it reads the data from a table stored in a Cassandra database and identifies what are the most commonly used adjectives for each of the major communication service providers (CSPs) in Brazil. After processing, the results are stored in another table in the same Cassandra database. Basically, it is a fancier version of the famous Hadoop word count example.

Unfortunately, there seem to be a lack of modern documentation about integrating Hadoop and Cassandra. Even the official guide seem to be deficient/outdated about this subject. To add insult to the injury, I also wanted to use composite keys, which complicated things further. After reading the example source code in Cassandra source code, I was able to successfully implement a working job.

Despite the lack of documentation and the hacking required to figure out how to make it work, the process is quite simple and even an unexperienced Cassandra/Hadoop developer such as myself can do it without much trouble. In the paragraphs below you will find additional details about the Hadoop and Cassandra integration and what is required to make it work.

Finally, as it's usual for my coding examples, the source code is available in my Github account under the open source Apache License v2.

First we need to setup the input configuration: it should be pretty simple, as you have to point to the Cassandra database instance, key space, input tables and the columns you'll be working with.

private void inputConfiguration(Configuration configuration) {
    ConfigHelper.setInputRpcPort(configuration, DB_PORT);
    ConfigHelper.setInputInitialAddress(configuration, DB_HOST);
    ConfigHelper.setInputPartitioner(configuration, PARTITIONER);

    ConfigHelper.setInputColumnFamily(configuration, KEYSPACE, INPUT_TABLE);

    // These are the columns we will be working with
    List<ByteBuffer> columns = Arrays.asList(
        ByteBufferUtil.bytes("reference_text"),
        ByteBufferUtil.bytes("domain"));

    SlicePredicate predicate = new SlicePredicate()
        .setColumn_names(columns);

    ConfigHelper.setInputSlicePredicate(configuration, predicate);
}

Then, we repeat the process for the output configuration. Since you don't have to setup the predicate or the input columns, it's even simpler than the input one:

private void outputConfiguration(Configuration configuration) {
    ConfigHelper.setOutputInitialAddress(configuration, DB_HOST);
    ConfigHelper.setOutputColumnFamily(configuration, KEYSPACE, OUTPUT_TABLE);
    ConfigHelper.setOutputPartitioner(configuration, PARTITIONER);

    String query = "UPDATE " + KEYSPACE + "." + OUTPUT_TABLE +
        " SET hash = ?, domain = ?, word = ?, occurrences = ?, reference_date = ? ";
    CqlConfigHelper.setOutputCql(configuration, query);
}

Configure the Hadoop job, with the appropriate classes and the input/output configuration

private Job getCSPWordJob(String[] args) throws IOException {
    Job job = new Job(getConf(), JOB_NAME);

    job.setJarByClass(Main.class);

    /**
    * This is related to what comes _out_ of the map process
    */
    job.setMapperClass(WordMapper.class);
    job.setReducerClass(CountReducerTable.class);

    job.setInputFormatClass(ColumnFamilyInputFormat.class);

    job.setMapOutputKeyClass(OccurrenceWritable.class);
    job.setMapOutputValueClass(IntWritable.class);

    job.setOutputKeyClass(Map.class);
    job.setOutputValueClass(ArrayList.class);
    job.setOutputFormatClass(ColumnFamilyOutputFormat.class);

    Configuration configuration = job.getConfiguration();

    inputConfiguration(configuration);
    outputConfiguration(configuration);

    return job;
}

Create your mapper class. To do so, extend Hadoop's Mapper class and create the map method. It may also be useful to create a getColumnValue method to simplify reading the value from the column.

public class WordMapper extends Mapper<ByteBuffer, SortedMap<ByteBuffer, Column>, OccurrenceWritable, IntWritable> {

    private final static IntWritable one = new IntWritable(1);

    /**
    * Given a column name, will return its value
    */
    private String getColumnValue(SortedMap<ByteBuffer, Column> columns, final String name) throws IOException, InterruptedException {
        ByteBuffer byteBuffer = Text.encode(name);
        Column column = columns.get(byteBuffer);
        String ret = ByteBufferUtil.string(column.value());

        return ret;
    }

    public void map(ByteBuffer key, SortedMap<ByteBuffer, Column> columns, Context context) throws IOException, InterruptedException {
        String referenceText = getColumnValue(columns, "reference_text");
        String domainText = getColumnValue(columns, "domain");

        /* Then you can run the map part of the job, saving whatever result you have to the current context */
        context.write(occurrenceWritable, one);
    }
}

Create your reducer job. To do so, you will have to extend the Hadoop's Reducer class and write the reduce method. In this class you'll write the reduce class that will save the results to the database.

public class CountReducerTable extends Reducer<OccurrenceWritable, IntWritable, ByteBuffer, ArrayList<Mutation>> {
    // Your code goes here
}

5.1 It may be useful to also create a few methods to create Mutation objects out of your data. The mutation objects are the ones that will actually be saved to the database. Here's an example:

/**
* Gets a mutation ...
* @param name The name of the table
* @param obj A string object
* @return A mutation object
*/
private static Mutation getMutation(String name, String obj) {
    org.apache.cassandra.thrift.Column c = new org.apache.cassandra.thrift.Column();

    // We really, really need to filter this, otherwise we save the
    // data with lots of invisible chars in the DB
    CharMatcher legalChars = CharMatcher.INVISIBLE;
    String filtered = legalChars.removeFrom(obj);

    c.setName(ByteBufferUtil.bytes(name));
    c.setValue(ByteBufferUtil.bytes(filtered));
    c.setTimestamp(System.currentTimeMillis());

    Mutation m = new Mutation();

    m.setColumn_or_supercolumn(new ColumnOrSuperColumn());
    m.column_or_supercolumn.setColumn(c);

    return m;
}

5.2 Finally, you can create the reduce logic in your reducer class. Basically, the process involves: a) run your reduce calculation, b) create the mutation objects that will be saved to the database, c) create the index to be saved to the database and d) write all of that to the context.

protected void reduce(OccurrenceWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
    int sum = 0;

    // First, iterate over the KV to calculate the count of references
    Iterator<IntWritable> it = values.iterator();
    while (it.hasNext()) {
        sum += it.next().get();
        String word = key.getText();
    }

    logger.info("Iterated over " + sum + " records for domain '" + key.toString() + "'");

    // Then, create the mutation objects so that we can save them on the DB
    Mutation domain = getMutation("domain", key.getDomain());
    Mutation word = getMutation("word", key.getText());
    Mutation occurrences = getMutation("occurrences", sum);
    Mutation date = getMutation("reference_date", (new Date()).getTime());

    // ... and put them in an array
    ArrayList<Mutation> list = new ArrayList<Mutation>();
    list.add(domain);
    list.add(word);
    list.add(occurrences);
    list.add(date);

    // ... since we cannot use composite keys very easily, calculates the
    // a SHA-1 hash to serve as the index. At the moment there's no need to
    // be fancy here, so we use a simple SHA-1 hash (index)
    String hashKey = DigestUtils.shaHex(key.getDomain() + key.getText());
    ByteBuffer b = ByteBufferUtil.bytes(hashKey);

    // ... and, finally, put it in the context
    context.write(b, list);
}