MapReduce & YARN

By the end of this lesson you will be able to explain the three phases of MapReduce, build a word count in plain Python that works the way Hadoop does, and describe how YARN gives each job its share of the cluster.

MapReduce is older than Spark, and few new projects write MapReduce code today. But Spark's jobs, stages and shuffles are direct descendants of it. Once you have seen word count done "by hand", Spark's execution plans will make sense.

The word-count story

Imagine the Lucknow city library wants to know the most common words in ten years of old newspapers. There are far too many pages for one person. So the head librarian, Meera, organises a team.

1. Map: each volunteer gets a stack of pages. For every word they read, they write a slip: "monsoon, 1", "cricket, 1", "monsoon, 1".
2. Shuffle: all the slips are sorted into boxes by word. Every "monsoon" slip goes to the same box, no matter which volunteer wrote it.
3. Reduce: one volunteer per group of boxes adds up the slips in each box and writes the final total: "monsoon, 15,210".

That is MapReduce. You write only two small functions, map and reduce. The framework handles splitting the input, moving the slips, sorting them, and recovering from failures.

Step 1: the map phase

The input is split into pieces called input splits, usually one per HDFS block. Each split goes to a map task. The map function turns each input line into key-value pairs.

splits = [
    "Rain in Pune today. Rain again tomorrow",
    "Cricket match in Pune cancelled due to rain",
    "Kochi expects rain and a cricket holiday",
]

def mapper(line):
    for word in line.lower().replace(".", "").split():
        yield (word, 1)

for i, split in enumerate(splits):
    pairs = list(mapper(split))
    print(f"map task {i}: {pairs[:5]} ... ({len(pairs)} pairs)")

map task 0: [('rain', 1), ('in', 1), ('pune', 1), ('today', 1), ('rain', 1)] ... (7 pairs)
map task 1: [('cricket', 1), ('match', 1), ('in', 1), ('pune', 1), ('cancelled', 1)] ... (8 pairs)
map task 2: [('kochi', 1), ('expects', 1), ('rain', 1), ('and', 1), ('a', 1)] ... (7 pairs)

Each map task works alone. It never needs to talk to the other map tasks, so all of them can run at the same time on different machines.

Step 2: shuffle and sort

Now every pair with the same key must reach the same reducer. Hadoop decides the reducer for a key with a hash: hash(key) % number_of_reducers. Then each reducer sorts its pairs by key. Together, this is the shuffle and sort phase. It moves data over the network, so it is the slowest part.

Step 3: reduce, all together

The full simulation below runs map, shuffle and reduce with two reducers. A simple, repeatable hash (the sum of the letter codes) stands in for Hadoop's hash function.

from collections import defaultdict

splits = [
    "Rain in Pune today. Rain again tomorrow",
    "Cricket match in Pune cancelled due to rain",
    "Kochi expects rain and a cricket holiday",
]
REDUCERS = 2

def mapper(line):
    for word in line.lower().replace(".", "").split():
        yield (word, 1)

def partition(key):
    return sum(ord(c) for c in key) % REDUCERS

# map + shuffle: send every pair to its reducer's box
boxes = [defaultdict(list) for _ in range(REDUCERS)]
for split in splits:
    for word, one in mapper(split):
        boxes[partition(word)][word].append(one)

# reduce: each reducer sums its own keys, in sorted order
def reducer(word, ones):
    return word, sum(ones)

for r, box in enumerate(boxes):
    results = [reducer(w, ones) for w, ones in sorted(box.items())]
    print(f"reducer {r}:", [x for x in results if x[1] > 1])

reducer 0: [('pune', 2), ('rain', 4)]
reducer 1: [('cricket', 2), ('in', 2)]

Each reducer printed only words that appear more than once, to keep the output short. "rain" appears four times across three map tasks, and all four slips met in reducer 0 because the partition function always sends "rain" to the same place.

Combiners: add up early

In the story, a volunteer who wrote "rain, 1" three times could write "rain, 3" once instead. That saves paper and carrying. In MapReduce, this local pre-aggregation is called a combiner. It runs on the map side and reduces the data sent through the shuffle.

from collections import Counter

splits = [
    "rain rain pune rain cricket rain",
    "rain cricket cricket pune rain rain",
]
without = sum(len(s.split()) for s in splits)
with_combiner = sum(len(Counter(s.split())) for s in splits)
print("pairs shuffled without combiner:", without)
print("pairs shuffled with combiner:   ", with_combiner)

pairs shuffled without combiner: 12
pairs shuffled with combiner:    6

Half the data moved. On real data with millions of repeated keys, the saving is much larger. A combiner only works for operations like sum, count, min and max, where partial results combine correctly (the same idea as in Distributed Computing Basics). Spark does this automatically for many aggregations.

Writing MapReduce in Python: Hadoop Streaming

Real MapReduce jobs are usually Java. Hadoop Streaming lets you write the mapper and reducer as any program that reads lines from standard input and prints lines to standard output. The reducer receives its lines already sorted by key.

# mapper.py
import sys

for line in sys.stdin:
    for word in line.lower().split():
        print(f"{word}\t1")

# reducer.py: input arrives sorted by word
import sys

current, count = None, 0
for line in sys.stdin:
    word, n = line.rstrip("\n").split("\t")
    if word != current and current is not None:
        print(f"{current}\t{count}")
        count = 0
    current = word
    count += int(n)
if current is not None:
    print(f"{current}\t{count}")

You submit it with the streaming JAR that ships with Hadoop:

hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \
  -files mapper.py,reducer.py \
  -mapper "python3 mapper.py" -reducer "python3 reducer.py" \
  -input /data/news -output /data/news-wordcount

You can test the same scripts on your laptop with a Linux pipe: cat news.txt | python3 mapper.py | sort | python3 reducer.py. The sort command plays the role of the shuffle.

YARN: sharing the cluster

A cluster runs many jobs at once: a MapReduce job, a Spark job, a Hive query. YARN (Yet Another Resource Negotiator) decides who gets which CPU cores and memory.

When you submit a job, the ResourceManager starts its ApplicationMaster in a container. The ApplicationMaster asks for more containers and runs tasks in them. Spark can run on YARN in exactly this way, with its executors living inside YARN containers. You will see this in Spark Architecture.

Why Spark replaced most MapReduce

MapReduce writes results to disk after every map and every reduce. A job that needs ten steps becomes ten MapReduce jobs, each reading and writing HDFS. This is slow for machine learning and interactive analysis, which repeat steps many times. Spark keeps intermediate data in memory where possible and plans many steps as one job, so the same work often runs much faster.

Common mistakes

• Doing heavy work in the reducer for a very common key. One reducer gets all the data for that key and becomes the bottleneck. This is called skew.
• Using a combiner for averages. Averages of averages are wrong; send sums and counts.
• Forgetting that reducer input is sorted. The streaming reducer above depends on all lines for one word arriving together.

Key points

• MapReduce has three phases: map, shuffle and sort, and reduce.
• The map function emits key-value pairs; all pairs with the same key reach the same reducer.
• A partition function such as hash(key) % reducers chooses the reducer.
• Combiners pre-aggregate on the map side and shrink the shuffle.
• YARN's ResourceManager, NodeManagers and ApplicationMasters share the cluster in containers.
• MapReduce writes to disk between steps; Spark keeps data in memory, so it is faster for multi-step work.

from collections import defaultdict

splits = [
    "Rain in Pune today. Rain again tomorrow",
    "Cricket match in Pune cancelled due to rain",
    "Kochi expects rain and a cricket holiday",
]
cities = {"pune", "kochi", "jaipur"}
REDUCERS = 3

def mapper(line):
    for word in line.lower().replace(".", "").split():
        if word in cities:
            yield (word, 1)

boxes = [defaultdict(list) for _ in range(REDUCERS)]
for split in splits:
    for word, one in mapper(split):
        boxes[sum(ord(c) for c in word) % REDUCERS][word].append(one)

for r, box in enumerate(boxes):
    print(f"reducer {r}:", [(w, sum(v)) for w, v in sorted(box.items())])

reducer 0: []
reducer 1: [('kochi', 1)]
reducer 2: [('pune', 2)]