Unit 3 — MapReduce and High-Level Tools

Topics Covered

Syllabus topics: MapReduce programming model: map, shuffle, reduce phases. Writing MapReduce applications in Hadoop. High-level abstractions: Hive, Pig, Crunch, and introduction to Spark integration.

3.1 The MapReduce programming model

map    : (k1, v1)        -> list of (k2, v2)
shuffle: gather every (k2, v2) with the same k2 onto one reducer
reduce : (k2, list of v2) -> list of (k3, v3)

⚠️ Map output goes to LOCAL DISK, not HDFS

This is examined and widely got wrong. Map output is intermediate — it is not replicated, not visible in HDFS, and deleted when the job ends. If a map task's node dies before its output is fetched, the map task is re-run, which is cheap. Replicating intermediate data would triple the cost of every job for no benefit.

3.2 The shuffle is the only expensive step

3.3 Combiners

⚠️ When a combiner is NOT safe

Demonstrated rather than asserted: groups [1,1,1,10] and [10] give a mean-of-means of 6.6250 against a true mean of 4.6000.

To average safely, emit (sum, count) pairs and divide only in the reducer. That is the same average-of-averages trap Business Intelligence Tools met in DAX, in a different costume — and Hadoop will not warn you.

3.4 Partitioning and skew

Hadoop's default partitioner is (hash(key) & MAXINT) % numReduceTasks.

With 3 reducers on the word-count data, the partitions hold [20, 17, 11] records — a largest/smallest ratio of 1.82 on 26 keys.

⚠️ Skew, not volume, is what kills a MapReduce job

Hash partitioning is only as balanced as the key distribution. Natural language is Zipfian, so a real corpus skews far worse: one reducer gets the and finishes long after the rest. The job's wall clock is that reducer.

Fixes, in order of preference:

1. Change the key. Prefix a salt to the hot key and aggregate twice.
2. A custom partitioner that splits known hot keys across reducers.
3. A combiner, which shrinks the hot key's volume before it moves.
4. More reducers — which helps least, because the hot key still lands on exactly one of them.

3.5 Writing MapReduce applications in Hadoop

The canonical Java is in WordCount.java and InvertedIndex.java, both run on the lab's cluster — lab.md, experiments 7 and 8, shows what they printed. Three details are examinable.

public void map(LongWritable key, Text value, Context context)

private final static IntWritable ONE = new IntWritable(1);
private final Text word = new Text();

⚠️ The reduce Iterable can be walked ONCE

for (IntWritable val : values) { sum += val.get(); }

values is streamed from disk. Calling values.iterator() a second time yields nothing — the classic bug when someone tries to compute both a sum and a count in one reducer. Accumulate everything you need in the single pass.

The inverted index adds one more

The filename is in neither the key nor the value. It comes from the input split:

FileSplit split = (FileSplit) context.getInputSplit();
fileName = split.getPath().getName();

And the inverted index cannot use a combiner, because the reducer's output type (a posting string) differs from its input type (a document id) — a combiner must have the same input and output types as the reducer.

3.6 Hive

⚠️ The partitioning trap

Partition on something with FEW distinct values. Partitioning by date on a small table makes four directories for nine rows — the small-files problem from Unit 2, created on purpose.

Rule of thumb: partition by quarter or month; bucket by customer id.

Bucketing

CLUSTERED BY (store) INTO 3 BUCKETS puts rows into files inside a partition by a hash of the column. Two tables bucketed the same way on the same column can be joined bucket to bucket with no shuffle — a sort-merge bucket join, and the reason bucketing exists.

Measured, and worth noting honestly: three stores over three buckets left bucket 1 empty. Hashing does not distribute small key sets evenly, and an empty bucket is still a file the job opens.

Managed against external tables

Use EXTERNAL for anything you did not produce and cannot recreate. A DROP TABLE on a managed table sitting over the company's only copy of a dataset is the classic Hive accident.

3.7 Pig

sales    = LOAD 'sales.csv' USING PigStorage(',') AS (...);
priced   = FOREACH sales GENERATE *, qty * list_price AS revenue;
bulk     = FILTER priced BY qty >= 6;
by_cat   = GROUP bulk BY category;
totals   = FOREACH by_cat GENERATE group, SUM(bulk.revenue);
STORE totals INTO 'out';

The operator mapping

⚠️ GROUP in Pig produces a bag, not an aggregate

C = GROUP B BY category  ->  (Grocery, {4 tuples})
                             (Personal, {1 tuple})
                             (Stationery, {2 tuples})

The bag is the value; FOREACH … GENERATE turns it into numbers. Hive fuses the two steps; Pig keeps them apart, which is why Pig can do things to a group that SQL cannot express without a window function.

Join strategies, which is where Pig gets specific

joined = JOIN priced BY store, stores BY store USING 'replicated';

3.8 Crunch, and the abstraction ladder

Apache Crunch is a Java library giving typed PCollection pipelines over MapReduce or Spark — closer to the code than Pig, more structured than raw MapReduce. It is in the syllabus, it is effectively dead (last release 2017), and the honest thing to say is that Spark's RDD API occupies its niche and won.

The ladder is worth drawing:

Higher is not better. Higher is faster to write and harder to tune. Hive is the right default; you drop a level when the optimiser gets it wrong.

3.9 Spark, introduced

Unit 5 covers Spark integration properly. What belongs here is why it replaced MapReduce.

Practice problems

1. Write the map and reduce functions for computing the average temperature per city, and explain why the obvious combiner is wrong.

map(record)    -> (city, temperature)
reduce(city, temps) -> (city, sum(temps) / len(temps))

The obvious combiner — the reducer itself — is wrong, because mean of means is not the mean unless every group is the same size.

The fix:

map(record)          -> (city, (temp, 1))
combine(city, pairs) -> (city, (sum of temps, sum of counts))
reduce(city, pairs)  -> (city, total_sum / total_count)

Divide only in the reducer. The combiner and reducer now have the same input and output types, which is also the condition Hadoop requires.

2. A word count over 1 TB of text produces a job where 99 reducers finish in 2 minutes and one runs for 3 hours. Diagnose and fix it.

Diagnosis: key skew. One reducer holds a stop word — almost certainly the — whose posting count is a large fraction of the corpus.

Fixes, in order:

1. Drop stop words in the mapper. Correct here, since nobody wants a count of the, and it removes the problem rather than managing it.

2. Salt the hot keys: emit (the#0, 1) … (the#9, 1) from the mapper, aggregate, then sum the ten partials in a second job.

3. A custom partitioner routing known hot keys to dedicated reducers.

What does not work: raising the reducer count. The hot key still lands on exactly one reducer.

3. You have a 2 TB fact table and a 40 MB dimension table to join. What do you do, and what would go wrong with the default?

Use a map-side (broadcast/replicated) join. The 40 MB dimension is loaded into every mapper's memory and the join happens with no shuffle at all.

• Pig: USING 'replicated'
• Hive: SET hive.auto.convert.join = true (on by default in modern Hive)
• Spark: broadcast(smallDf)

The default reduce-side join would shuffle all 2 TB across the network so that matching keys meet — for a table that fits in RAM on every node.

The failure mode: if the "small" table is actually 4 GB, every mapper tries to hold 4 GB and the job dies with OutOfMemoryError. Broadcast joins have a size threshold for a reason.

4. Why does Hive have no indexes, and what replaces them?

Hive removed indexes in version 3 because they were expensive to maintain and rarely faster than a full scan on a columnar format. What replaces them:

• Partition pruning — skip whole directories
• Bucketing — skip files, and enable shuffle-free joins
• Columnar formats (ORC, Parquet) — read only the columns needed
• Row-group statistics — skip blocks whose min/max cannot match
• Vectorised execution — process a thousand rows per call

Together these do what an index does, without a structure to keep in sync — and the underlying reason is that analytical queries touch a large fraction of rows, where an index loses to a scan anyway.

5. Trace word count on ["a b a", "b c"] through all five phases with two reducers.

INPUT SPLITS   split1 = "a b a"          split2 = "b c"

MAP            (a,1) (b,1) (a,1)         (b,1) (c,1)

COMBINE        (a,2) (b,1)               (b,1) (c,1)
   per task, so split1's two a's merge and split2's cannot help

PARTITION      hash(a)%2, hash(b)%2, hash(c)%2  -- say a,c -> R0; b -> R1

SHUFFLE        R0 gets (a,2) (c,1)
               R1 gets (b,1) (b,1)

REDUCE         R0: a=2, c=1             R1: b=2

Exam questions from this unit

Two marks

1. State the map and reduce signatures.
2. Where does map output go?
3. What is a combiner?
4. Give one aggregate for which a combiner is unsafe.
5. What is the default partitioner?
6. What does ILLUSTRATE do?
7. Give one difference between a managed and an external Hive table.
8. Why does Spark beat MapReduce on iterative jobs?

Five marks

1. Explain the five phases of a MapReduce job.
2. Explain combiners, with an example where one is unsafe.
3. Explain partitioning and skew, and three ways to fix skew.
4. Compare Hive and Pig, and say when you would choose each.
5. Explain partitioning and bucketing in Hive, with the trap in each.

Ten marks

1. Explain the MapReduce programming model in full, with word count traced through every phase, and discuss the shuffle's cost.

2. Compare MapReduce, Hive, Pig and Spark as ways of processing the same dataset, with the strengths and limitations of each.

Mistakes that cost marks