WordCount Example

Neptune lets users run and connect arbitrary programs (called "tasks") via the AppScale cloud platform. As such, one programming model that is easily supported via Neptune is map-reduce, in which one or more occurrences of a "map" program run, and connect their output to one or more "reduce" programs. Here, we discuss how Neptune can be used to run a workflow that counts the number of words in an input document (aka WordCount from the Google MapReduce paper).

First, we assume that you have Neptune installed on your computer - if you don't, run gem install neptune.

Next, we assume that you have AppScale running somewhere (e.g., EC2, Eucalyptus, Xen) - if you don't, get that running and come back here once you do!

Let's begin by writing a Map function that reads in the file that it's been assigned to read, splits each line in that file, and counts the number of occurrences of each word in its file. Here's a Python implementation:

map.py:

import json
import sys

def wc(f):
  words = {}
  for line in f:
    for word in line.split():
      try:
        words[word] += 1
      except KeyError:
        words[word] = 1
  return json.dumps(words)


if __name__ == "__main__":
  file_to_read = sys.argv[1]
  f = open(file_to_read)
  result = wc(f)
  print str(result)

Next, let's write a Reduce function that accepts the output of one or more Map functions, and for each file that the Map function produced, sums each individual total into a grand total (that is, the total count across all files). A Python implementation of that would look like the following:

reduce.py:

import json
import sys

def reduce(files):
  counts = {}
  for this_file in files:
    contents = json.loads(this_file)
    for k, v in contents.iteritems():
      try:
        counts[k] += v
      except KeyError:
        counts[k] = v
  return json.dumps(counts)

if __name__ == "__main__":
  map_contents = []
  for i, filename in enumerate(sys.argv):
    if i == 0:
      continue
    f = open(filename)
    map_contents.append(f.read())
  result = reduce(map_contents)
  print str(result)

Finally, let's connect it all together with a Neptune script. Our Neptune script will copy our Map and Reduce functions to Amazon S3 (so make sure you have EC2_ACCESS_KEY, EC2_SECRET_KEY, and S3_URL specified in your environment), as well as the input texts to parse through. Here we just have a single input text, the complete works of William Shakespeare (grabbed from Project Gutenberg). Next, our Neptune script starts as many Map tasks as we like, grabs their output locations, and feeds those in as inputs to the single Reduce task. We then fire off all the Map tasks, and once they finish, the Reduce task starts. Here's the same thing I just said, but in code form:

run_wc.rb:

NUM_MAPPERS = 5

def make_n_tasks(params)
  tasks = []
  NUM_MAPPERS.times { |i|
    tasks << params
  }
  return tasks
end

common_params = {
    :storage => "s3",
    :is_remote => false,
    :bucket_name => "your bucket name here",
    :run_local => true,
    :engine => "executor-sqs"
}

puts "Started at #{Time.now}"
puts "Starting Shakespeare Wordcount, with #{NUM_MAPPERS} map tasks and 1 reduce task"
STDOUT.flush

map_params = common_params.dup
map_params[:code] = "/home/cgb/map.py"
map_params[:argv] = ["/home/cgb/shakespeare.txt"]
map_tasks = babel(make_n_tasks(map_params))

outputs = []
map_tasks.each { |t|
  outputs << t.job_data['@output']
}

puts "done with map, starting reduce"
STDOUT.flush

reduce_params = common_params.dup
reduce_params[:code] = "/home/cgb/reduce.py"
reduce_params[:argv] = outputs
reduce_task = babel(reduce_params)

puts reduce_task.stdout

We invoke this script with neptune run_wc.rb (note we use the Ruby extension since Neptune is an extension of Ruby and thus the same syntax highlighting works for Neptune code), and we're good to go! Here's a few exercises to keep you going now that you're excited to mess around with Neptune:

• Although we call this map-reduce, don't confuse it with Google MapReduce. We didn't do any input splitting, shuffling, or sorting like MapReduce does. Implement any of these functions, to allow support for multiple Reduce functions.
• Since we don't split the input text beforehand, every Map function is getting an identical copy of the Shakespeare text. Run a separate Split task before the Map tasks are run that splits the input files into N files, where N = number of Map tasks to run.