MapReduce and YARN

What MapReduce isSearch on YouTubeBookmarkSaved

MapReduce is a framework for distributed and parallel processing of large data volumes. The idea comes from functional programming. It applies one function to many items independently. The map step transforms items. The reduce step collates items into new values. This pattern suits clusters well.

Dean and Ghemawat published the method in 2004. Their paper "MapReduce: Simplified Data Processing on Large Clusters" describes split, process and aggregate steps for the Google search engine. The open-source version shipped inside Apache Hadoop.

The slide pipeline has six stages:

1. Data block.
2. Read and process (map jobs).
3. Generate key-value pairs.
4. Receive key-value pairs (reduce side).
5. Shuffle and sort.
6. Aggregate key-value pairs into smaller sets.

MapReduce programming splits the input into independent chunks. Map tasks process the chunks fully in parallel. A reduce task then combines the map outputs into the final result.

Phases of MapReduceSearch on YouTubeBookmarkSaved

The four phases run in this order: split, map, shuffle and sort, reduce. Input chunks enter the map tasks. The map outputs are grouped by key. The reduce tasks emit the final pairs.

The slides draw the data change at each step. Input splits hold records. Map turns each record into intermediate key-value pairs. Grouping collects all values of one key into (k, [v, v, ...]). Reduce turns each group into output pairs.

Map phase and partitioningSearch on YouTubeBookmarkSaved

The map task reads its assigned input split from HDFS. It parses the split into records as key-value pairs. It applies the user map function to each record. Then it informs the JobTracker of completion.

An input split is not an HDFS block. A split is the MapReduce work unit. Splits often match blocks (Hadoop and HDFS explains blocks). The count rule is strict:

The partition phase rules from the slides:

• Each map task decides which reduce task receives each output.
• The destination partition is the same for any one key.
• The number of partitions equals the number of reduce tasks.

Shuffle, sort and reduceSearch on YouTubeBookmarkSaved

The shuffle phase fetches the map outputs for one partition from all map tasks. The sort phase merge-sorts them into a single run per key. The reduce phase applies the user reduce function to each group and emits the output.

The job execution diagram adds three more pieces. A RecordReader parses the split into records before the map task runs. A Combiner pre-aggregates map output on the map task node before the network shuffle. The output lands on HDFS through an OutputFormat.

Word count: sentence splitsSearch on YouTubeBookmarkSaved

The slide input is one paragraph: "Big data comes in various formats. This data can be stored in multiple data servers". It is split into two chunks. Each chunk gets one map task.

• Split 1: "Big data comes in various formats".
• Split 2: "This data can be stored in multiple data servers".

Map emits (word, 1) for each token. Split 1 emits 6 pairs. Split 2 emits 9 pairs (data appears twice).

Word count: Deer Bear RiverSearch on YouTubeBookmarkSaved

The slide input has three lines. Each line is one split, so the job runs 3 map tasks.

• Split 1: "Deer Bear River".
• Split 2: "Car Car River".
• Split 3: "Deer Car Bear".

The Endeavour paragraphSearch on YouTubeBookmarkSaved

The slides trace a longer document about the space shuttle Endeavour crew. The slide shows only the start of the trace. The map function emits (word, 1) per token: (The, 1) (crew, 1) (of, 1) (the, 1) (space, 1) (shuttle, 1) (Endeavor, 1) (recently, 1). The reduce function collects all values of each key. The slide shows five outputs: (crew, 2) (space, 1) (the, 3) (shuttle, 1) (recently, 1).

Hadoop v1: JobTracker and TaskTrackerSearch on YouTubeBookmarkSaved

Hadoop v1 runs two daemons for MapReduce. The JobTracker is the master daemon. It runs each MapReduce job. It links Hadoop and the application. It creates the execution plan: which task runs on which node. It monitors all running tasks. When a task fails, it reschedules the task on another node after fixed retries. One cluster runs exactly one JobTracker.

The TaskTracker runs the assigned tasks on its node. It sends heartbeat messages to the JobTracker. When the JobTracker misses heartbeats from a TaskTracker, it marks that node as failed. It resubmits the task to another available node.

The master runs the JobTracker and the NameNode; each slave runs a TaskTracker and a DataNode. A map task (also called mapper) runs a map function. A reduce task (also called reducer) runs a reduce function.

YARNSearch on YouTubeBookmarkSaved

YARN is Yet Another Resource Negotiator. It arrived in Hadoop 2.0. It is the middle layer between HDFS and MapReduce. It manages cluster resources: memory, network bandwidth, disk IO and CPU. The two basic resource types are memory and virtual cores (vcores).

YARN runs many engines, not only MapReduce:

YARN daemonsSearch on YouTubeBookmarkSaved

The ResourceManager is the global job manager. It has two components. The Scheduler allocates resources across all applications. It never monitors job status. The Applications Manager accepts job submissions. It restarts ApplicationMasters on failure.

A NodeManager is a per-node agent. It launches each application container. It monitors the resource use of every container. It tracks the jobs running on its node.

An ApplicationMaster is the per-application manager. It negotiates resource containers with the Scheduler. It tracks container status and job progress. It restarts failed tasks. A container is the resource unit: memory, CPU, disk and network for one task.

The slide legend names four message types: MapReduce status (solid), job submission (dashed), node status (dash-dot) and resource request (dotted).

Execution sequenceSearch on YouTubeBookmarkSaved

1. Submit the application from the client.
2. Ask the ResourceManager for one container to start the ApplicationMaster.
3. Register the ApplicationMaster with the ResourceManager at boot.
4. Negotiate further containers with the ResourceManager.
5. Contact the NodeManager to launch each granted container.
6. Run the application code inside the container and report the status.
7. Poll the ApplicationMaster or the ResourceManager for progress.
8. Unregister the ApplicationMaster and shut it down. Free its container.

Splitting the JobTracker fixes its bottleneck. No single daemon plans every task and tracks every job. New engines reuse the same resource layer.

Quick self-testSearch on YouTubeBookmarkSaved

1. A job has 4 input splits. How many map tasks run, and why?

4 map tasks. One map task handles each input split, so the task count equals the split count.

2. Letter sums are Bear 26, Car 22, Deer 32, River 72. With 3 reduce tasks, which partition holds each key?

Bear: $26 \bmod 3 = 2$. Car: $22 \bmod 3 = 1$. Deer: $32 \bmod 3 = 2$. River: $72 \bmod 3 = 0$. Bear and Deer share partition 2.

3. A split holds "Car Car River". How many pairs cross the network with and without a combiner?

Without: 3 pairs, (Car, 1) twice plus (River, 1). With: 2 pairs, (Car, 2) plus (River, 1). The final sums are equal.

4. Map outputs are (a, 1) (b, 1) (a, 1). State the shuffle groups and the reduce outputs.

Groups: a with (1, 1), b with (1). Reduce: (a, 2), (b, 1).

5. Differentiate the JobTracker and the TaskTracker.

The JobTracker plans jobs, monitors tasks and reschedules failures. The TaskTracker runs single tasks and heartbeats the JobTracker.

6. Why must the number of partitions equal the number of reduce tasks?

Each partition feeds exactly one reduce task. One key must land whole in one partition, so the counts match by construction.

7. Differentiate the Scheduler and the Applications Manager.

The Scheduler allocates resources across applications and never monitors status. The Applications Manager accepts submissions and restarts failed ApplicationMasters.

8. Why did YARN split the JobTracker into two daemons?

One JobTracker that plans every task and tracks every job becomes a bottleneck. YARN separates global scheduling (Scheduler) from per-application management (ApplicationMaster), and so lets Tez, Storm and other engines share the cluster.

9. A NodeManager stops heartbeating and an ApplicationMaster crashes. Who restarts what?

The ResourceManager stops sending work to the silent node. The Applications Manager restarts the crashed ApplicationMaster. The ApplicationMaster restarts its own failed tasks in fresh containers.

10. In the sentence-split example, what are the shuffle group and reduce output for data?

Group data, (1, 1, 1) and output 3. The word occurs once in split 1 and twice in split 2.