The chapter contains a lot of details about integer numbers encoding and compression. Since these topics are not directly about MapReduce, I made no questions about them.
4.4 Inverting Indexing: Revised Implementation
Explain inverting index retrieval algorithm. You may assume that each document fits into the memory. Assume also then there is a huge number of documents. Which part should be optimized by integer encoding and compression?
Input: text documents
key: document id
value: text document
Output: key/value pairs where
key: word
value: list[documentId, numberOfOccurences]
list elements must be sorted by numberOfOccurences
Mapper counts number of occurrences in the document for each word. As the whole document fits into the memory, we can hold partial results in a map.
Intermediate key/values:
key: word, numberOfOccurences
value: documentId
Custom partitioner groups intermediate key/values by word. Custom sort sorts them primary by word and secondary by the number of occurrences.
Reducer uses initialize method to initialize list of all postings. Reduce method handles two cases:
- current word equal to previous word – add documentId and numberOfOccurences to posting list.
- current word equal to previous word – emit previous word and posting list; initialize posting list.
Posting list in reducer should be compressed.
class MAPPER
method INITIALIZE
H = new hash map
method MAP(docid, doc d)
H = new hash map
for all term w in doc d do
H(w) = H(w) + 1
for all term w in H do
emit(pair(u, w), count 1)
method CLOSE
for all term w in H
emit(pair(w, H(w)), docid)
class REDUCER
variable previous_word = 0
variable PL = new list of postings
method REDUCE(pair (w, #occurrences), docid)
if w <> previous_word && previous_word <> 0 do
emit(w, PL)
PL = new list of postings
PL.add(pair(#occurrences, docid))
previous_word = w
method compare(key (w1, o1), key (w2, o2))
if w1 = w2
return keys are equal
return keys are different
class SORTING_COMPARATOR
method compare(key (w1, o1), key (w2, o2))
if w1 = w2 do
return compare(o1, o2)
return compare(w1, w2)
5 Graph Algorithms
The chapter contains two algorithms: shortest path in the graph and page ranking algorithm. The questions are straightforward.
5.2 Parallel Breadth-First Search
Find shortest path from one node origin to all other nodes. Each edge has a weight associated. Input key/value pairs have already bean preprocessed into comfortable form.
Input: graph
key: node id
value: distance to origin, list[adjacent node, edge length]
Output: key/value pairs where
key: node id
value: distance to origin, list[adjacent node, edge length]
The algorithm requires multiple iterations. It stops the iteration does not change any ‘distance to origin’. At worst, there will be O(n) iterations where n is a number of nodes in the graph.
Mapper passes original graph to the next iteration as it is. Plus, it generates key/value pair for each adjacent node. The value contains the minimum known distance from origin if the route would go through node.
class MAPPER
method MAP(node, pair(dist, adjacencylist))
emit(node, pair(dist, adjacencylist))
for all (closenode, nodedist) in adjacencylist do
emit(closenode, pair(dist + nodedist, empty))
Reducer finds the minimum known distance from each node. It passes the distance along with the original graph to the next iteration. It also increments global counter whenever minimum known distance to any node changes.
class REDUCER
method REDUCE(node, list(dist, adjacencylist))
minimum = infinity
previous_iteration_solution = infinity
original_graph = empty
for all (dist, adjacencylist) in list do
if adjacencylist not empty do
original_graph = adjacencylist
previous_iteration_solution = dist
if minimum > dist
minimum = dist
if previous_iteration_solution <> minimum
increment global counter
emit(node, pair(minimum, original_graph))
If the global counter is 0, the algorithm stops. Otherwise another iteration is needed.
Explain page rank algorithm, assume alpha = 0.
Page rank P(n) of a page n is calculated form page ranks of all pages linking to it.
P(n) = sum_m (P(m)/C(m))
The sum goes through all pages m linking to the page n. C(m) is the number of outgoing links of the page m.
Page rank algorithm runs in iterations. Mapper passes page rank contribution of each page to adjacent pages. Reducer updates page rank of each node. The algorithm stops when page ranks no longer moves.
class MAPPER
method MAP(page, (page_rank, adjacency_list))
emit(page, (0, adjacency_list))
contribution = page_rank/adjacency_list.length
for all node in adjacency_list do
emit(node, (contribution, empty))
class REDUCER
method REDUCE(page, contributions[c1, c2, ..., cn])
rank = 0
adjacency_list = new list
for all c in contributions do
adjacency_list.addAll(c.adjacency_list)
rank = rank + c.contribution
emit(page, (rank, adjacency_list))
6 EM Algorithms For Text Processing
I made no questions out of this chapter.
Exercises
This chapter contains hands-on exercises for MapReduce. Some of them require multiple iterations.
Warm Up
Count number of occurrences of every word in a text collection.
Input:
key: document id,
value: text document.
Output:
key: word,
value: number of occurences.
Intermediate pairs:
key: word
value: integer - how many times was the word seen in the input.
class MAPPER
method MAP(docid a, doc d)
for all term w in doc d do
emit(w, 1)
class COMBINER
method COMBINE(word w, counts[c1, c2, ..., cn])
s = 0
for all c in counts[c1, c2, ..., cn] do
s = s + c
emit(word w, s)
class REDUCER
variable total_occurrences = 0
method REDUCE(word w, counts[c1, c2, ..., cn])
s = 0
for all c in counts[c1, c2, ..., cn] do
s = s + c
emit(word w, s)
Alternative solution would use in-mapper combining.
Web Store
Website user log contains user ids and length of each session. The website has modest number of registered users. Compute the average session length for each user.
Input:
key: user id,
value: session length.
Output:
key: user id,
value: average session length.
As the number of registered users is modest, we can use in-mapper combining.
class MAPPER
variable total_time = new hash map
variable sessions_number = new hash map
method MAP(user_id, session_length)
total_time(user_id) = total_time(user_id) + session_length
sessions_number(user_id) = sessions_number(user_id) + 1
method CLOSE
for all user_id in total_logged_in_time
tt = total_time(user_id)
sn = sessions_number(user_id)
emit(user_id, pair(tt, sn))
class REDUCER
method REDUCE(user_id, [pairs(time, sessions_number)])
total_time = 0
total_sessions = 0
for all pairs in [pairs(time, sessions_number)] do
total_time = total_time + time
total_sessions = total_sessions + sessions_number
emit(user_id, total_time/total_sessions)
Web store log contains user id and bought item for each sale. You need to implement “buyers of item also bought” functionality. Whenever the item is shown, the store will suggest five items most often bought by items buyers.
Input:
key: user id,
value: brought item.
Output:
key: item,
value: list of five most common "buyers of item also bought" items.
Our solution has two iterations. First iteration generates lists of all items brought by the same user. Grouping is done by the framework, both mapper and reducer perform an identity function.
Input:
key: user id,
value: brought item.
Output:
key: user id,
value: list of all brought items.
class MAPPER
method MAP(user_id, item)
emit(user_id, item)
class REDUCER
method REDUCE(user_id, items[i1, i2, ..., in])
emit(user_id, items)
Second iteration solves co-occurrences problem on list items. It uses the stripes approach. Only difference against the standard solution is that we have emit only five most common co-occurrences.
Input:
key: user id,
value: list of all brought items.
Output:
key: item,
value: list of five most common co-occurrences.
class MAPPER
method MAP(user_id, items[i1, i2, ..., in])
for all item in items do
H = new hash map
for all item j in items do
H(j) = H(j) + 1
emit(item, H)
class REDUCER
method REDUCE(item, stripes[H1, H2, ..., Hn])
T = new hash map
for all H in stripes do
for all (key/value) in H do
T(key) = T(key) + value
emit(user_id, max_five(T))
Web store log contains user id, timestamp, item and number of brought pieces for each sale. The store is looking for items whose sales rise or decline at the same time. Find 20 item couples with maximum of such months.
Input:
key: user id,
value: timestamp, brought item, count.
Output:
key: item, item
value: number of months when both items sales rose or decline.
#: the output contains 20 key/value pairs with maximum value
Our solution requires multiple MapReduce iterations. We have to:
- calculate whether items sales for any given month went up or down,
- create lists of items with the same sales change during the same month,
- find number of co-occurrences in those lists,
- choose items with maximum co-occurrences.
First iteration calculates sales changes for any given month. We have to supply mapper, partitioner, custom sort and reducer. Mapper generates one intermediate key/value pair for each input key/value. Key is composed of sold item and sales month. Value contains number of sold pieces.
Partitioner sends all key/value pairs with the same item to the same reducer. Custom sort sorts them by months. Finally, reducer calculates sales changes.
Input:
key: user id,
value: timestamp, item, count.
Intermediate key/values:
key: item, month
value: count.
Output:
key: month, up/down/equal
value: item.
class MAPPER
method MAP(user_id, (timestamp, item, count))
month = get_month(timestamp)
emit((item, month), count)
class PARTITIONING_COMPARATOR
method compare(key (item1, month1), key (item2, month2))
if item1 = item2
return keys are equal
return keys are different
class SORTING_COMPARATOR
method compare(key (item1, month1), key (item2, month2))
if item1 = item2 do
return compare(month1, month2)
return compare(item1, item2)
class REDUCER
method REDUCE((item, month), counts[c1, c2, ..., cn])
c = sum([c1, c2, ..., cn])
if last_item = item
if last_month + 1 = month
//emit correct up/down/equal flags
if last_count < count
emit((item, month), up)
if last_count > count
emit((item, month), down)
if last_count = count
emit((item, month), equal)
else
//no sales during some months
emit((item, last_month + 1), down)
emit((item, month), up)
else
// new item
emit((last_item, last_month + 1), down)
emit((item, month), up)
last_item = item
last_count = count
last_month = month
Second iteration groups first iteration results by keys. It generates lists of items with same sales changes during the same month. Framework does all the work. Both mapper and reducer perform an identity function.
Input:
key: month, up/down/equal
value: item.
Output:
key: month, up/down/equal
value: [items].
Third iteration performs standard ‘co-occurrences by pairs’ algorithm.
Input:
key: month, up/down/equal
value: [items].
Intermediate key/values:
key: item, item
value: partial number of co-occurrences.
Output:
key: item, item
value: number of months when both items sales rose or decline.
#: the output contains all items couples
class MAPPER
method MAP((month, change), items[i1, i2, ..., in])
for each i in items do
for each j in items do
if i != j
emit((i, j), 1)
class COMBINER
method COMBINE((item1, item2), co-occurrences[c1, c2, ..., cn])
s = 0
for all c in co-occurrences[c1, c2, ..., cn] do
s = s + c
emit((item1, item2), s)
class REDUCER
method REDUCE((item, item), co-occurrences[c1, c2, ..., cn])
s = 0
for all c in co-occurrences[c1, c2, ..., cn] do
s = s + c
emit((item1, item2), s)
Finally, we have to choose 20 key/value pairs with maximum value. Each mapper selects 20 key/value pairs with maximum value and emits them with the same key. There will be only one reducer which selects final 20 key/value pairs.
Input:
key: item, item
value: number of months when both items sales rose or decline.
#: the output contains all items couples
Intermediate key/values:
key: 1
value: item, item, number of months when both items sales rose or decline.
#: the output contains 20 key/value pairs with maximum value for each mapper
Output:
key: item, item
value: number of months when both items sales rose or decline.
#: the output contains 20 key/value pairs with maximum value
the code is very simple but long
Criminal Agency
Inputs to all exercises in this chapter uses the same data structure.
Criminal agency stole Facebook’s friendships database and wants to analyze new data. Friendships are stored in form key/value pairs, each friendship corresponds to two key/value pairs:
Friends:
key: first friend name
value: second friend name
key: second friend name
value: first friend name
The agency owns also criminal records of all citizens:
Criminal record:
key: citizen name
value: when, where, accomplices, description
Find at risk youths. A person is considered at risk youth if more than half of his friends have criminal record.
Our solution has two iterations. First iteration joins two sets and flags each ‘value friend’ with has_record/law_abiding flags.
Output:
key: first friend
value: second friend, has_record/law_abiding
The mapper flags each key with data set name. Partitioner groups data according to names in keys and sorter puts criminal records before friendships. We could use local aggregation to remove multiple criminal records for the same person.
class MAPPER
method MAP(name, value)
if value is name do
emit(name, friendship, item)
else
emit(name, criminal, item)
class PARTITIONING_COMPARATOR
method compare(key (name1, dataset1), key (name2, dataset2))
if name1 = name2
return keys are equal
return keys are different
class SORTING_COMPARATOR
method compare(key (name1, dataset1), key (name2, dataset2))
if name1 = name2 AND dataset1 is criminal
return key1 is lower
if name1 = name2 AND dataset2 is criminal
return key2 is lower
return compare(name1, name2)
class REDUCER
variable previous_name
method REDUCE(pair(name, flag), items[i1, i2, ..., in])
if flag is criminal do
previous_name = name
has_record = criminal
return
if previous_name <> name do
has_record = law_abiding
else
has_record = criminal
previous_name = name
for all i in items do
emit(i.name, pair(name, has_record))
Second iteration counts both total number of friends and number of friends with criminal record. Reducer emits key/value pairs only for at risk youths. Also this iteration could use some kind of local aggregation.
Intermediate key/value:
key: name
value: total friends, total friend criminals
# totals are relative only to in data sets subsets
Output:
key: name
value: empty
# only at risk youths
class MAPPER
method MAP(name, pair(name, has_record))
if has_record is law_abiding do
emit(name, pair(0, 1))
else
emit(name, pair(1, 1))
class REDUCER
method REDUCE(name, items[pair(total, criminals)])
total = 0
criminals = 0
for all i in items do
total = total + i.total
criminals = criminals + i.criminals
if criminals / total > 0.5 do
emit(name, empty)
Find gangs. Gang is a group of people that:
- has exactly 5 members,
- each member is friend with all other members,
- each two members committed at least 3 crimes together.
Again, we need three iterations. The idea is to first clean up the graph of all useless edges, so that only criminal contacts remain. Then, we split graph into smaller manageable sub-graphs. We attach all criminal contacts and edges between them to each person:
Last iteration reducers input:
key: person
values: all his criminal contacts and relationships between them.
Final reducer takes smaller graphs represented by value in each key/value pair and finds complete sub-graphs with 4 vertices in it. Add person from the key in it, and you have found a complete sub-graph with 5 vertices. The reducer may use any polynomial algorithm.
First iteration uses pairs approach to clear the graph. We omit both local aggregation and removal of duplicities. Both would make the algorithm more efficient.
Intermediate key/values:
key: first friend, second friend, friendship/accomplice
value: 1
Output:
key: first friend, second friend
value: empty
# only friends with common criminal record
class MAPPER
method MAP(name, value)
if value is name do
emit(triple(name, value, friendship), empty)
else
for all crime_accomplice in value.accomplices do
emit(triple(name, crime_accomplice, accomplice), 1)
class PARTITIONING_COMPARATOR
method compare(key (name1, accomplice1, flag1), key (name2, accomplice2, flag2))
if name1 = name2 AND accomplice1 = accomplice2
return keys are equal
return keys are different
class SORTING_COMPARATOR
method compare(key (name1, accomplice1, flag1), key (name2, accomplice2, flag2))
if name1 = name2 AND accomplice1 AND flag1 is friendship
return key1 is lower
if name1 = name2 AND accomplice1 AND flag2 is friendship
return key2 is lower
return compare(pair(name1, accomplice1), pair(name2, accomplice2))
class REDUCER
variable previous_name
variable previous_accomplice
method sameAsPrevious(name, accomplice)
if previous_name <> name
return false
if previous_accomplice <> accomplice
return false
return true
method REDUCE(triple(name, accomplice, flag), items[i1, i2, ..., in])
if sameAsPrevious(name, accomplice) do
if items.length > 2 do
emit(name, accomplice)
return
if flag is friendship do
previous_name = name
previous_accomplice = accomplice
Second iteration attaches lists of all ‘second degree’ friends to edges:
Input
key: first friend, second friend
value: empty
Intermediate key/values:
key: first friend
value: first friend, second friend
key: second friend
value: first friend, second friend
Output:
key: first friend, second friend
value: all friends of second friend
key: second friend, first friend
value: all friends of first friend
class MAPPER
method MAP((first friend, second friend), empty)
emit(first friend, (first friend, second friend))
emit(second friend, (first friend, second friend))
class REDUCER
method REDUCE(name, edges[e1, e2, ..., en])
friends = new Set
friends.add(name)
for all edge in edges do
friends.add(edge.v1, edge.v2)
for all edge in edges do
emit(edge, friends)
Finally, mapper and shuffle and sort phase together generate lists of all friends of any given person and relationships between them.
Input
key: friend 1, friend 2
value: all friends of friend 2
Intermediate key/values:
key: friend 1
value: friend 2, all friends of friend 2
Reducers input (after shuffle and sort):
key: person
values: all his friends and relationships between them.
Output:
key: first friend, second friend, third friend, fourth friend, fifth friend
value: gang
class MAPPER
method MAP((friend , friend 2), all friends of second friend)
emit(friend 1, (friend 2, all friends of friend 2))
class REDUCER
method REDUCE(name, graph attached to it)
any polynomial algorithm will work
Reference: MapReduce Questions and Answers from our JCG partner Maria Jurcovicova at the This is Stuff blog.
