A simple Hadoop job using Clojure

12 September, 2012

This is a re-post of an original post I wrote on Tumbler in 2012, which for now is accessible at its original URL: https://perdalum-blog.tumblr.com/post/31543901149/a-simple-hadoop-job-using-clojure.

At work we’ve been experimenting with MapReduce on the Hadoop platform. That means a lot of Java stuff, tinkering with both programming in Java and cluster administration. All of it rather exiting, albeit a bit frustrating at times. Luckily, we’ve now reached a point where the Cloudera Manager is used for the cluster administration. Our first cluster consists of five 12-core Intel Blade servers each with about 50 GB of RAM and 4 TB of HDFS storage. The number of servers will probably be doubled in the coming months, so we’ll actually have some pretty nice hardware to experiment with.

There exists a lot of different introductions to the MapReduce programming model, so I won’t go into that here. Instead, I’ll just give you how I got introduced to it. I’ve been enjoying programming in functional languages for some years, primarily LISP and Mathematica. That interest gave a rather natural angle to MapReduce, like, yes off course that’s the way to do it. Though it took me some time to get from the ordinary functional Map and Reduce/Apply model to the key, value-based approach of MapReduce. In Mathematica, Python, LISP, etc., it’s practical impossible to do any data manipulation without mapping and reducing your data.

Still, there’s a long way from

data = Apply[reducer, Map[mapper[#], list]]

to the standard word count Java example in MapReduce Tutorial. I really hate boiler plate code. I hate having to write the same word twice. Off course there’s a lot of ways to avoid that in Java, but I like another approach to overcome that, so here enter Clojure and I’m about ready to show my little Hadoop job written in Clojure but before commencing the show, I’ll start by showing you the code of my-counter — Go grab it and cast some Leinigen spells!

The complete Hadoop job:

(ns my-counter.core
  (:require [clojure-hadoop.wrap :as wrap]
            [clojure-hadoop.defjob :as defjob]
            [clojure-hadoop.imports :as imp])
  (:import (java.util StringTokenizer))
  (:use clojure-hadoop.job))

(imp/import-io)
(imp/import-mapreduce)

(defn my-map [key value]
  (map (fn [token] [token 1]) (enumeration-seq  (StringTokenizer. value))))

(defn my-reduce [key values-fn]
  [[key (reduce + (values-fn))]])

(defn string-long-writer [^TaskInputOutputContext context ^String key value]
  (.write context (Text. key) (LongWritable. value)))

(defn string-long-reduce-reader [^Text key wvalues]
  [(.toString key)
   (fn [] (map (fn [^LongWritable v] (.get v)) wvalues))])

(defjob/defjob job
  :map my-map
  :map-reader wrap/int-string-map-reader
  :map-writer string-long-writer
  :reduce my-reduce
  :reduce-reader string-long-reduce-reader
  :reduce-writer string-long-writer
  :output-key Text
  :output-value LongWritable
  :input-format :text
  :output-format :text
  :compress-output false
  :replace true)

I not yet quite sure what every s-exp in the Clojure files do, but I like what I see, how concise the code is. The intention is very easy to read, and the boiler plating is kept to a minimum.

Apart from the implementation of the actual job I also like the project structure and the iterative development is facilitates. You can run the job and test it without a Hadoop system! Just execute the command:

$ java -cp target/my-counter-0.1.0-SNAPSHOT-standalone.jar clojure_hadoop.job -job my-counter.core/job -input shakespeare.txt -output out

On the Hadoop cluster the corresponding command is:

$ hadoop jar my-counter-0.1.0-SNAPSHOT-standalone.jar clojure_hadoop.job -job my-counter.core/job -input shakespeare.txt -output out

The next step will be to actual run the code at the REPL in a real explorative programming style.