Weeks 3-4: MapReduce

Question 1

Distributed File System (DFS)

Answer 1

This system stores large data, provides replication, and protects against failures. It’s used when files are huge (100’s of GB+), existing files are rarely updated, and reads and appends are common.

Examples: Google GFS, Hadoop HDFS, CloudStore

Question 2

Software Stack

Answer 2

It uses a DFS and a programming model, which does large-scale, distributed computations efficiently, as well as tolerate hardware failures.

Question 3

Chunk Servers

Answer 3

DFS’s divide files into chunks and store them in chunk servers, which are replicated on different machines or racks (to protect from server or rack failures).

Question 4

Master Node

Answer 4

It’s a metadata file used to find chunks of files. The master node is also replicated. The directory for the file system is also replicated.

Question 5

Client Library

Answer 5

It accesses files and communicates with the master node, as well as connects to chunk servers.

Question 6

Hadoop HDFS

Answer 6

The Hadoop Distributed File System (HDFS) is the primary storage system used by Hadoop.

Question 7

HDFS Read Path

Answer 7

1. Obtain the locations of the blocks. The client is requesting the block locations before reading the file.
2. Check if the file exists.
3. Check if the client has read permission.
4. Sort the block (chunk) locations by distance to the client.
5. The datanodes (chunk servers) stream data. If the server becomes unavailable, the client can read another replica of the file on a different server.

Question 8

HDFS Write Path

Answer 8

1. Create the file request.
2. Check if the file exists.
3. Check if the client has write permission.
4. The outputstream object is created.
5. Split the data into packets and add them into the queue.
6. The namenode (master node) to allocate new blocks (chunks)
7. Make connection with datanodes (chunk servers).
8. Datanode 1 consumes data from the queue, writes to Datanode 2 (for replication), which sends acknowledgement to Datanode 1,…, Datanode 1 sends the data back to the client.
9. The client closes the outputstream and sends the request to close the file. This is when the master node is updated.

Question 9

MapReduce

Answer 9

This is a programming model for handling large datasets.

Workflow:
1. The input data is split into pieces and multiple instances of the programme begin.
2. One instance is elected as the master, and the rest are workers.
3. A worker with a map task processes its inputs, outputs key-value pairs, and passes them into the user’s map function. Pairs are buffered in the memory.
4. Buffered pairs are written to the local disk and partitioned into regions. The locations of pairs on the disk are passed to the master.
5. When a reducer worker is notified by the master about locations, it reads buffered pairs, and groups them by key (key, list-of-values).
6. The reducer worker passes the key and values into the user’s reducer function, whose output is appended to the final output file.
7. When all map and reduce tasks are complete, control returns to the user programme. The programme may use output as input to another MapReduce task.

Question 10

Map Task

Answer 10

The purpose of this task is to extract something of interest. It obtains chunks from the DFS, converts them into key-value pairs according to the user-written code.

Question 11

Shuffle and Sort Task

Answer 11

This is done by the system automatically. Data from all mappers are grouped by key, split among reducers, and sorted by key. Each reducer gets all values associated with the same key.

Question 12

Reduce Task

Answer 12

This task aggregates, summarises, filters, or transforms the data. Each reducer obtains all values associated with the same key. Code tells what key-value pairs to create.

Question 13

Map Function

Answer 13

Input: key-value pair
Output: a list of key-value pairs (can be empty)
One map function calls for every different key-value pairs.

Question 14

Reduce Function

Answer 14

Input: key-value pair
Output: a list of key-value pairs (can be empty)
One reduce function called for every different keys.

Question 15

WordCount

Answer 15

The objective of WordCount is to return key-value pairs, with the keys being words and the values being the counts of the words.

Input: text file
1. The text file is parsed, so that there is one key-value pair for each word.
2. The key-value pairs are grouped by key.
3. The key-value pairs are reduced, so that there’s one key-value pair for each unique word, with its corresponding count.

Note that the key-value pairs are ordered by how early the first instance of the word appears in the text file.

Question 16

Master Failure

Answer 16

In this case, the MapReduce task is aborted and must restart.

Question 17

Map Worker Failure

Answer 17

The map tasks completed or in-progress at the worker are reset to idle. Reduce workers are notified when the task is rescheduled to another worker.

Question 18

Reduce Worker Failure

Answer 18

Only in-progress tasks are reset to idle. The reduce task is rescheduled to start later.

Question 19

Map to Reduce Ratio

Answer 19

Usually, there are many more map tasks than reduce tasks. This improves dynamic load balancing, speeds up recovery from worker failures, and can pipeline shuffling with map execution. Usually, the number of reduce tasks is a small multiple of the number of nodes, as output is spread across reduce files.

Question 20

Backup Tasks

Answer 20

This deals with the issue of slow workers, which may be caused by multiple jobs on the same machine, bad disks, and bugs. The solution is to spawn backup copies of in-progress tasks whenever a MapReduce operation is about to complete. Whichever backup copy that finishes first wins and is used. This dramatically shortens job completion time.

Question 21

Combiners

Answer 21

Often the map task will make many pairs from the same key (i.e. WordCount). Combiners save network time by pre-aggregating values in the mapper. It’s usually the same as the reduce function, but only works for functions that are both commutative and associative.

Question 22

Partitioner

Answer 22

This divides the data (key, value) pairs to Map tasks. Partitioners ensure that the same key, which can be output by multiple mappers, goes to the same reducer.

By default: hash(key) mod R

This aims to create “well-balanced” partitions. It’s sometimes useful to override the hash function.

Example: hash(hostname(URL)) mod R puts all URL’s from the host in the same output file.

Question 23

mrjob

Answer 23

This is a Python library specific for MapReduce jobs.

Question 24

WordCount Code

Answer 24

File: mr_word_count.py

from mrjob.job import MRJob

class MRWordFrequencyCount(MRJob):

def mapper(self, _, line):
yield “chars”, len(line)
yield “words”, len(line.split())
yield “lines”, 1

def reducer(self, key, values):
yield key, sum(values)

if __name__ == ‘__main__’:
MRWordFrequencyCount.run()

Note that mapper has an empty key and a line as value, that mapper returns 3 key-value pairs, and for each key, returns the sum of its values.

Question 25

Map-Reduce Join

Answer 25

If you need to join tuples of form (a,b) and (b,c) into the form (a,c), then the map process turns each input tuple R(a,b) into key-value pair (b,a(R)) and each input tuple S(b,c) into key-value pair (b,(c,S)).

Question 26

Communication Cost

Answer 26

This cost measure sums up all the I/O of all processes. Often, communication is slower than computation, at least for simple algorithms. This is because simple computations still require lots of reading and writing of data.

Communication cost = input file size + 2 x (sum of the sizes of all files passed from Map processes to Reduce processes) + the sum of the output sizes of the Reduce processes

Question 27

Elapsed Communication Cost

Answer 27

This cost measure finds the maximum of the I/O along any path.

Question 28

Elapsed Computation Cost

Answer 28

This cost measure is similar to elapsed communication cost, but counts running time for processes as well. It functions as a wall-clock time using parallelism.

Elapsed communication cost = the largest input + output (for any map process) + the largest input + output (for any reduce process)

Question 29

Cost Measure

Answer 29

Either the I/O (communication) or the processing (computation) cost dominates. The total cost indicates how much cost there is to use a cloud service.

Question 30

Key Complexity

Answer 30

This is a complexity measure for MapReduce. Over all key-value pairs, input to or output by any Mapper or Redicer, calculate the maximum size, maximum run time, and maximum memory used.

Question 31

Sequential Complexity

Answer 31

This is a complexity measure where over all mappers and reducers, you sum up the size of all key-value pairs input and output by Mapper and Reducers, and the total running time for all Mappers and Reducers.

Question 32

Master MapReduce Instance

Answer 32

This instance is established as master by election, with the rest of the MapReduce instances becoming workers. The master MapReduce Instance periodically pings workers. It also assigns reduce tasks to idle works.

Question 33

MapReduce Drawbacks

Answer 33

1. It’s slower than alternatives such as Spark.
2. It’s based off Hadoop, which is inefficient because it writes to a distributed file system.
3. It’s inefficient for applications that reuse intermediate results across multiple computations (i.e. PageRank, k-means clustering).
4. It’s inefficient for multiple ad-hoc queries on the same data (this is common in data mining).