Let's get started with some examples using
Octopython, which is our lightweight
MapReduce implementation that we shall be
using for our programming assignments
later this week.
Let's take a look at how a word count
problem can be expressed in Octopython.
This is a slight modification of the file
which you'll find on the site as a demo
example.
The directory Gutenburg Small contains a
bunch of small text files much smaller
than the ones which are available on the
site so that we can actually execute this
program in a few seconds rather than a few
hours.
Notice that we have defined two functions,
the map and the reduce.
The map function takes a key value pair
and emits other key value pairs.
And so does the reduce.
Let's see what the map function is doing.
The key, in this case, is the document ID,
or the document name, the file name here.
And the value is the contents of that
file.
For every line in that file, and for every
word in that line.
The map function emits a pair which is the
word itself in lower case and the count of
one.
The reduce function gets a bunch of key
value pairs as well.
In this case, the keys are the words and
the values are lists of counts that it's
getting from many different mappers.
The reduce function also gets a bunch of
key value pairs from all the mappers.
In this case, the keys are words, and the
values are the counts which are just the
ones emitted by the mappers.
And so, for every word, all the reduce
function needs to do is figure out how
many mappers found that word how many
times.
And all it needs to do is figure out the
length of the list of values it gets from
all the mappers for a particular word.
Notice that the map functions are emitting
counts of one.
So it's okay for the reducer to merely sum
up the lists of values that it gets from
all the mappers.
We'll now run this MapReduce program using
Octo.
This terminal is the server.
So when we start it by simply calling the
Octo code with the key word, server.
And the program that implements the map
reduced functions which is our program
here.
A server gets started waiting for clients
to join.
We have two more shells where we will
start clients by merely calling Okto with
the client keyword, and telling it which
server to connect to.
That is which machine the server is
running on.
The server in our case is running on the,
very machine itself, so we use localhost,
as the address, for the server.
Let's start each of the appliance and as
you can see, they are performing different
map tasks on different files and then
finally they cooperate to conform all the
reduce tasks, in the end writing the
totals to a file as well as to the screen.
So, for example they have computed that
across all these files, there are four
occurrences of you, two occurrences of
yet, and many others which are scrolled up
beyond our view.
The rest of code is really used by the
server to set things up.
In particular a source dictionary needs to
get created with the key value pairs that
each mapper needs to get are kept in a
dictionary.
So here we have the file name as the key
and the file contents as the value for all
the files in the directory Gutenberg
Small.
Notice also that both clients and servers
get the same program.
It's just that the clients implement the
map and reduce operations and the server
merely gets data from where it needs to
find it and sends it out to the mappers
and finally accumulates the data from the
reducers and prints out the results.
When you run this code for yourselves some
of you will notice that the mappers or
rather the processors in the map phase are
only implementing mapper operations and
emitting counts of one, rather than, as we
have described earlier performing partial
reduced operations from whatever documents
that they manage to get data for.
Lets see how much data is actually
produced by the map phase because of this
behavior.
Each word is emitted by a mapper multiple
times.
In fact if the size of the initial set of
documents is d, including every occurrence
of every word, after processing by
mappers, again, exactly that much data is
produced, because each word is emitted
every time it is seen.
On the other hand, in our earlier
implementation or description of the word
counting problem this was not the case.
Since the counts within each map process
were actually summed up before moving on
to the reduced phase.
This process of reduce operations being
performed partially within each mapper is
called a combined phase.
In other words, when you sum up the word
counts for every mapper before emitting,
you're implementing the reduce function as
a combined function before the data get
sorted by the MapReduce system.
The result of this is a reduction in the
amount of intermediate data which is
produced which will lead to better
parallel efficiency of the map produced
process as we'll see shortly.
See if we can modify our Octo program to
implement a combined phase where partial
reduce operations are performed before
passing the data onto the actual reduce
phase.
Unfortunately Octo doesn't allow for a
separate combine operation like the map
and reduce functions.
However, we can simulate the combined
process by ensuring that each time a math
function is called.
It keeps track of its results in a local
variable W.
The reason this is possible, is because of
a peculiar feature in Python, called the
iterator.
Where a function that instead of returning
a value, yields a value.
Stays in memory and all it's variable
defined remain available next time the
function is called with exactly the same
set of inputs.
And the function begins implementing
exactly where it left off last time.
So let's see how it works.
Every time the map function is called for
a particular document, it adds up the word
counts for all the words that it sees in
that document, and stores them in this
dictionary W.
Then it goes thru all the words that its
found in this dictionary with counts, and
yields them as key value pairs.
When the yield operation is called, the
function actually returns this value to
the caller, which is the Octo program.
The next time the Octo program asks for a
value, it yields the next count, and so on
and so forth.
In a sense, this is implementing the
combiner within the map function.
A little bit of cheating, because in real
map reduce implementations we actually
have a separate combine function, in
addition to the map and the reduce
functions.
>> Let's now run our modified combiner
enabled map reduce task.
We start the server and then we start each
client in turn as you can see they are
performing the map and then the reduced
phase and now we are done.
You might not have noticed earlier, but
the server is actually printing out the
time, the combined MapReduce task actually
took.
In this case, it was twelve seconds.
The actual Octo program available from the
site doesn't have this timing.
So I would encourage you to try to look
through that code and insert the timing,
so that you can reproduce such results.
Further, do look back and figure out how
much time the.
Word count program without the combiner
actually took and you'll see that the
combiner phase actually added a lot of
efficiency.
Last but not least it's very important to
realize that the combined function can
work only if the reduce function is a
commutative and associative one because we
don't know which order these reduce
functions are being performed in on what
data.
So, any reduce function which relies on
the fact that the function is applied in a
particular order on all the values for a
particular key will not be able to be
split up into a combined phase for
efficiency purposes.