Google's MapReduce as Groovy closures

Google’s MapReduce programming model [http://labs.google.com/papers/mapreduce.html] serves for processing and generating large data sets. Users specify a mapper function that processes a key/value pair to generate a set of intermediate key/value pairs, and a reducer function that merges all intermediate values associated with the same intermediate key. Many real world tasks are expressible in this model, as shown in the paper. Here we illustrate two simple examples: (a) for counting occurrences of words in documents; and (b) for creating an inverted index of the words in documents.

Both examples use the same mapReduce function, implemented as a Groovy closure. The function accepts two parameters representing, respectively, the mapper and reducer functions (also closures). The mapper, reducer and mapReduce functions exploit many of the functional programming techniques supported through Groovy closures. To aid the development we have prepared an abstract class ListFunctions which defines many of the standard list processing functions (closures) such as map, filter, composition, etc. The abstract class MapFunctions provides similar functions when applied to Maps.

Example 1: Word count

import com.adt.fp.ListFunctions as LF
import com.adt.fp.MapFunctions as MF

def mapReduce(mapper, reducer) {
def reducePerKey = LF.composition.curry(MF.mapWithKey.curry({ k, v -> v}),
LF.composition.curry(MF.filterWithKey.curry({ k, v -> (v != null)}),
MF.mapWithKey.curry(reducer)))

def groupByKey = { kvList ->
def insert = { mp, kv -> return MF.insertWith(LF.join, kv[0], [kv[1]], mp) }
return LF.foldL(insert, MF.empty(), kvList)
}

def mapPerKey = LF.composition.curry(LF.concat,
LF.composition.curry(LF.map.curry(LF.uncurry(mapper)), MF.toList))

return LF.composition.curry(reducePerKey, LF.composition.curry(groupByKey, mapPerKey))
}

def flip = { f, x, y -> return f(y, x) }
def mkPair = { x, y -> return [x, y] }

def tokenize = { str -> return str.tokenize() }
def sum = LF.foldL.curry({ x, y -> x + y }, 0)

def occurMapper = { key, words -> return LF.map(flip.curry(mkPair, 1), tokenize(words)) }
def occurReducer = { key, occurs -> sum(occurs) }

def wordOccurrenceCount = mapReduce(occurMapper, occurReducer)

def docs = ['doc1': 'fold the fold', 'doc2': 'appreciate the unfold']

assert wordOccurrenceCount(docs) == ["unfold":1, "the":2, "appreciate":1, "fold":2]

Example 2: Inverted index

import com.adt.fp.ListFunctions as LF
import com.adt.fp.MapFunctions as MF

def mapReduce(mapper, reducer) {
// …
}

def flip = { f, x, y -> return f(y, x) }
def mkPair = { x, y -> return [x, y] }

def tokenize = { str -> return str.tokenize() }

def indexMapper = { key, words -> return LF.map(flip.curry(mkPair, key), tokenize(words)) }
def indexReducer = { key, docs -> docs.unique().sort() }

def invertedIndex = mapReduce(indexMapper, indexReducer)

def docs = ['doc1': 'fold the fold', 'doc2': 'appreciate the unfold']

assert invertedIndex(docs) ==
["unfold":["doc2"], "the":["doc1", "doc2"], "appreciate":["doc2"], "fold":["doc1"]]

The mapReduce function returns the function (closure) that performs the required processing. It is defined in terms of its mapper function parameter and its reducer function parameter. The definition for mapReduce is in terms of its support functions mapPerKey, groupByKey and reducePerkey [http://www.cs.vu.nl/~ralf/MapReduce/]. The helper function mapPerKey exports the Map input data to a List then maps the mapper function over this List. The support function groupByKey groups intermediate values by intermediate keys. Finally, reducePerKey maps the reducer function over the groups of intermediate data.

In Example 1, the mapper function makes the pairs [word, 1] for every word in the source input. It is implemented by mapping the function flip.curry(mkPair, 1) over each word. This clever function uses mkPair, to produce the pair and flip ensures they are in the correct order. The reducer function for this example simply sums each of the 1s in the intermediate values.

In Example 2, the mapper function delivers the pairs [documentedID, word] for every word in the source input. The reducer function removes any duplicates and sorts the List of documentIDs from the intermediate values.

Sample code and a more expansive decription can be found at [Groovy MapReduce].

Combinator Parsers and External DSLs

For the past few months I have been investigating the opportunities and challenges of Domain Specific Languages (DSL) in Groovy. One issue surrounding external DSLs (eDSL) is the need to compile the DSL into code that accesses the application API. This often suggests parser generators such as javaCC or Antlr. If they are unfamiliar to the developer they can present a steep learning curve. They also require that additional code for the DSL has to be developed in addition to the application code.

Another consideration would be to use a parser combinator framework such as Jparsec. This is a classic Java implementation of a combinator parser exhibiting all the usual OO characteristics. It is also a very large library with many classes, many methods, etc.

While seeking a solution to some of these issues I also sought one that would exhibit some of the strengths of Groovy. The approach described in [Combinator Parser] exploits Groovy closures. Each closure defines a simple combinator parser that can be combined into more elaborate parsers. The parser is then transformed with yet another combinator into a compiler [External Domain Specific Languages]. Writing a parser/compiler is then no more complex than writing assignments that reflect the DSL grammar. For example:

def factor = altC(identifier, integer)  // factor: identifier  integer
def term = sepByC(factor, multiply)     // term: factor { '*' factor }*
def expr = sepByC(term, add)            // expr: term { '+' term }*

Here, the altC combinator parsers alternatives, while the sepByC parses one or more occurences of the first parser separated by the second parser. The sepByC combinator is, in fact, an assembly of the more primitive combinators seqC (a sequence of two parsers) and noneOrMoreC (none or more occurrences of a parser).

Each combinator is defined as a method that returns a closure (the parser). For example, the satisfyC combinator appears as:

def satisfyC = { pred ->
    return { inp -> if(pred(inp[0])) ... success ... else ... fail ... }
}

The closure determines if the first item in the input stream satisfies some predicate (a closure with a single parameter, returning a boolean). Observe how the predicate is a 'free' variable in the closure definition and is set by the call to satisfyC. Examples might include:

def isLetter = { ch -> ... is ch a letter? ... }
def isDigit = { ch -> ... is ch a digit? ... }
def letter = satisfyC(isLetter)
def digit = satisfyC(isDigit)

from which it is easy to define:

def identifier = seqC(letter, noneOrMoreC(altC(letter, digit)))
def integer = oneOrMoreC(digit)

Pleasingly, each combinator is only a few lines of Groovy code [ Combinators.groovy]. The complete set is defined in less than 200 LOC! The combinators are generic insofar as the input stream may be a character sequence (a character parser) or a token stream ( a parser).

Currently, I am working to extend the code with support for error reporting. Additionally, I have defined a combinator parser that reads a BNF definition and generates the combinator parser for that grammar - a kind of combinator parser compiler compiler.

I should be pleased to receive feedback from other developers.

Ken Barclay