Thinking in MapReduce

1 The story

Counting votes the way India does

The college wants one number per item: how many orders and how much money, for every canteen on every campus, for five years. Riya's Python script finishes the 4.7 MB file of one canteen in a tenth of a second. Then Kabir shows her the real folder: 30 campuses, five years, one file per day. Several terabytes. Her laptop would need days just to read it, and the files are spread over hundreds of DataNodes anyway.

Riya

Sir, can't I just copy all the files to one big server and run my script?

Arjun Sir

Think of election counting day. Crores of votes are not carried to Delhi. Every counting table counts its own machines, the totals for each candidate are collected, and then added. MapReduce is the same three steps: map where the data lives, shuffle the partial results so each key ends up in one place, and reduce each key to its answer.

Kabir

And the part that really matters: you only write the map and the reduce. Hadoop moves the data between them, runs your code on every node, and reruns whatever fails.

Arjun Sir

Today we write our first real MapReduce job, run it on a cluster with HDFS and YARN, and read what Hadoop tells us about it: 1 map task for the small file, 4 for the big one, and one task that Hadoop quietly ran twice.

2 Definition

MapReduce

Formal definition (write this in exams)

MapReduce is a programming model and an execution framework for processing large data sets in parallel on a cluster. The programmer supplies two functions: a map function, applied independently to every input record, that turns a key-value pair (k1, v1) into zero or more intermediate pairs list(k2, v2); and a reduce function that receives each intermediate key with all its values, (k2, list(v2)), and produces the output list(k3, v3). Between them the framework performs the shuffle and sort: it partitions the intermediate pairs by key, transfers them to the reducers and sorts them so all values of a key arrive together. The framework also splits the input, schedules map tasks close to their data (data locality), re-executes failed tasks and writes the output to a distributed file system. In Hadoop 2 and later, MapReduce jobs run on YARN, with one MRAppMaster per job.

In simple words

You describe what to do with one record (map) and how to combine everything that shares a key (reduce). Hadoop runs the map on every block in parallel, groups the results by key, and runs the reduce once per key.

Words you'll meet in this lesson

JobOne complete MapReduce program run: input, map, shuffle, reduce, output.

TaskA unit of work: one map task per input split, one reduce task per partition.

Task attemptOne execution of a task. A task can have several (retries, speculation).

Key-value pairThe only data shape MapReduce knows, e.g. ("chai", 15).

Intermediate dataMap output: written to the worker's local disk, not HDFS.

ShuffleMoving and sorting map output so each key reaches one reducer.

MRAppMasterThe per-job manager on YARN that asks for containers and tracks tasks.

Speculative executionRunning a backup attempt of a slow task; the first to finish wins.

From our lab (Hadoop 3.5.0, HDFS + YARN, 3 workers)

Small input	`canteen_orders.csv`, 4.7 MB: 1 split → 1 map + 1 reduce, 300,001 input records, 6 output lines, 25 s in total. Plain Python on one file: 0.1 s
Bigger input	`orders_x90.csv`, 422 MB (2.7 crore orders): 4 splits → 4 maps on worker1, worker2 (2) and worker3, all data-local; job 68 s
Speculation	Map 0 was slowest (43.9 s), so Hadoop started a copy on worker3; the original won and the copy was KILLED: “Launched map tasks=5, Killed map tasks=1”
Shuffle	27,000,000 map output records, 324 MB moved to the single reducer, which saw 6 groups (one per item)

MapReduce at three levels

Pick your level. New to this? Start at Beginner and climb.

Beginner Counting day

Every counting table (a worker) counts only the machines on its own table: that is map. Nobody carries all votes to one hall.
Partial totals are sorted by candidate and sent to the right desk: that is the shuffle.
Each desk adds the partial totals of one candidate: that is reduce. Ten tables or a thousand, the method doesn't change.
If one table's counter falls ill, another counter recounts that table only. Hadoop does the same with a failed task.

Intermediate What you write, what Hadoop does

You write: a Mapper (map(key, value, context)), a Reducer (reduce(key, values, context)) and a small driver that sets input, output and classes.
Hadoop does: splits the input (Lesson 8.2), starts one map task per split on a node that holds the block, sorts and spills map output, copies it to reducers, merges and groups by key (Lesson 8.4), writes part-r-NNNNN files and _SUCCESS.
Map tasks never talk to each other, and neither do reduce tasks. That independence is what lets Hadoop run them anywhere and rerun any one of them.
For our job: map turns “1002,vada pav,20” into (“vada pav”, 20); reduce turns (“vada pav”, [20, 20, …]) into vada pav 49986 999720.

Advanced Thinking in keys

The design question is always: what is the key? Everything that must be combined has to share a key, because only values with the same key ever meet.
Map output is the expensive part: it is sorted, written to local disk, and sent over the network. Emit as little as possible (filter early, use compact types, add a combiner in Lesson 9.1).
Reducers see each key once, with its values in no guaranteed order. Anything that needs order inside a key needs a secondary sort (Lesson 9.7).
MapReduce fits batch jobs that scan everything once. Iterative work (machine learning, graph algorithms) and interactive queries pay the disk write between steps on every pass, which is why Spark and Tez took over those jobs.

Exam tip: For “Explain MapReduce with an example”, write the two signatures map(k1, v1) → list(k2, v2) and reduce(k2, list(v2)) → list(k3, v3), name the shuffle and sort in between, and show one record passing through all three (e.g. “1002,vada pav,20” → (vada pav, 20) → vada pav 49986 999720).

3 Types and ways to use it

Common MapReduce patterns

Most MapReduce programs are one of a handful of patterns. The difference is what the map emits as key and value.

Pattern	Map emits	Reduce does	Example
Summarization	(group, number)	sum, count, min, max	Orders and revenue per item (this lesson)
Filtering	the record, or nothing	none (map-only job)	Keep valid orders only (Lesson 8.3)
Distinct	(value, nothing)	write each key once	Unique student IDs that ordered this month
Inverted index	(word, document id)	list of documents per word	Search: which feedback lines mention “cold”
Join	(join key, tagged record)	combine records of both sides	Orders + menu prices (Lesson 9.5)
Top-N	local top N per map	global top N	Top 10 items by revenue

The pieces of a MapReduce job on YARN

Client

Computes the splits, uploads the jar and configuration to HDFS, submits the application to the ResourceManager. Our log: “number of splits:4”.

your terminal

MRAppMaster

Runs in the first container. Requests one container per task, preferring nodes that hold each split, and tracks progress, failures and speculation.

one per job

Map and reduce tasks

Each runs in its own container (a YarnChild JVM) on a NodeManager. Our lab: map containers of 384 MB.

many per job

JobHistory Server

Keeps finished jobs: counters, every task attempt, where it ran and how long it took.

after the job

Big Data Fundamentals Lesson 3.1 and Hadoop Lesson 3.2 showed the life of a job from far away. This module opens it up, one stage per lesson.

The same model lives on in newer tools. A Hive GROUP BY or a Spark reduceByKey still does map, shuffle by key, reduce; they just keep more in memory and chain steps without writing to HDFS in between.

4 Step by step

Our first MapReduce job, on a real cluster

This module uses a new lab cluster with YARN: a NameNode, a ResourceManager, a JobHistory Server and three workers, each running a DataNode and a NodeManager, so map tasks can run where their blocks are. The setup is in “Run it on your PC”.

Write the map and the reduce

// map: one CSV line in, one (item, amount) pair out
String[] f = line.toString().split(",");
if (f.length != 3 || f[0].equals("order_id")) return;
item.set(f[1]);  amount.set(Integer.parseInt(f[2]));
ctx.write(item, amount);

// reduce: one item and ALL its amounts in, one line out
for (IntWritable a : amounts) { orders++; revenue += a.get(); }
ctx.write(item, new Text(orders + "\t" + revenue));

Neither function knows how big the file is or how many machines exist. That is the whole point: the same code runs on 4.7 MB or 4.7 TB.

Submit the job
```
hadoop jar mr.jar ItemRevenue /user/hadoop/data/canteen_orders.csv /out/item_revenue
```
INFO FileInputFormat:303 - Total input files to process : 1 INFO JobSubmitter:203 - number of splits:1 INFO YarnClientImpl:356 - Submitted application application_1792601400460_0001 INFO Job:1754 - map 0% reduce 0% INFO Job:1754 - map 100% reduce 0% INFO Job:1754 - map 100% reduce 100% INFO Job:1765 - Job job_1792601400460_0001 completed successfully
The client computed one split (the file is a single 4.7 MB block), uploaded the jar, and YARN started an MRAppMaster, one map container and one reduce container.
Read the counters
```
Launched map tasks=1
Launched reduce tasks=1
Data-local map tasks=1
Map input records=300001
Map output records=300000
Reduce input groups=6
Reduce output records=6
```
300,001 lines in (one is the header, which the map skipped), 300,000 pairs out of the map, 6 distinct keys at the reducer, 6 lines written. Counters are the first thing to check when a result looks wrong.
Look at the output
```
hdfs dfs -ls /out/item_revenue
hdfs dfs -cat /out/item_revenue/part-r-00000
```
/out/item_revenue/_SUCCESS /out/item_revenue/part-r-00000 chai 50170 752550 dosa 49948 2247660 idli 49984 1499520 poha 49891 1247275 samosa 50021 900378 vada pav 49986 999720
One file per reducer, named part-r-00000, and an empty _SUCCESS marker. The keys come out sorted because the shuffle sorted them.
Feed it 90 times more data
```
hadoop jar mr.jar ItemRevenue /user/hadoop/data/orders_x90.csv /out/item_revenue_x90
bash attempts.sh job_1792601400460_0003
```
number of splits:4 Killed map tasks=1 Launched map tasks=5 Data-local map tasks=5 attempt_…_m_000000_0 SUCCEEDED worker2 43.9 s attempt_…_m_000000_1 KILLED worker3 11.7 s Speculation: attempt_…_m_000000_0 succeeded first! attempt_…_m_000001_0 SUCCEEDED worker3 30.6 s attempt_…_m_000002_0 SUCCEEDED worker1 29.1 s attempt_…_m_000003_0 SUCCEEDED worker2 28.6 s attempt_…_r_000000_0 SUCCEEDED worker1 25.1 s
Four blocks, four map tasks running at the same time on three workers, each next to its block. Map 0 lagged, so the MRAppMaster launched a speculative copy on worker3; the original finished first and the copy was killed. Same code, no change.
Be honest about speed
```
4.7 MB  plain Python, one process        0.1 s
4.7 MB  MapReduce job on 3 workers       25 s
403 MB  one Java process reading HDFS    5.9 s   (Lesson 6.5)
422 MB  MapReduce job, 4 maps            68 s
```
On a laptop-sized cluster, MapReduce is slower: starting containers, sorting, writing map output to disk and copying it costs tens of seconds. It wins when the data is far bigger than one machine can read in time: 10 TB at 200 MB/s is 14 hours for one machine, but under 2 minutes for 500 machines reading their own blocks in parallel.

The output folder must not exist. Hadoop refuses to start a job whose output directory is already there, so it can never overwrite results by accident. Delete it first (hdfs dfs -rm -r /out/item_revenue) or write to a new folder.

5 See it happen

One job across three workers

Six steps of the 422 MB job: submit, map where the blocks are, a speculative copy, shuffle, reduce, output.

Map, shuffle, reduce 6 steps

What you see

Behind the screen

Try it yourself: a job in the playground

Upload a file, run a word count job, and compare a run with and without data locality.

6 Real-world use

MapReduce at work

1. Building a search index

MapReduce was created at Google to rebuild its web index: map emits (word, page) for every page, reduce builds the list of pages per word. Hadoop's first users (Nutch, Yahoo!) did the same.

Crawled pages→map: (word, page)→reduce: posting lists

2. Telecom and bank batch jobs

Nightly jobs summarise billions of call detail records or card transactions per customer: a classic summarization with the customer number as key.

Records (HDFS)→key = customer→Totals per customer

3. Log analytics

Web and app logs from thousands of servers land in HDFS; jobs count errors, users and page views per hour. Facebook and Yahoo! ran tens of thousands of such jobs a day in the 2010s.

Logs→key = (hour, page)→Dashboards

4. Under SQL engines

Hive originally compiled every SQL query into MapReduce jobs; a GROUP BY item is our job. Today Hive runs on Tez and most new work uses Spark, but the map → shuffle → reduce shape is the same.

GROUP BY→map + shuffle + reduce→Result table

Simplified; organisations differ in tools and policies.

7 Run the code

Simulate map, shuffle and reduce

This is the same job as ItemRevenue.java, written as three plain Python steps so you can see the data between them. Hadoop does the shuffle for you; here we do it with a dictionary.

lines = ["order_id,item,amount", "1001,samosa,18", "1002,vada pav,20", "1003,poha,25",
         "1004,samosa,18", "1005,chai,15", "1006,poha,25", "1007,chai,15", "1008,chai,15"]

def map_fn(line):                       # one record in, zero or more (key, value) pairs out
    f = line.split(",")
    if f[0] == "order_id":
        return []
    return [(f[1], int(f[2]))]

def reduce_fn(item, amounts):           # one key with ALL its values in
    return (item, len(amounts), sum(amounts))

mapped = [pair for line in lines for pair in map_fn(line)]
print("map output :", mapped[:4], "...")

groups = {}                             # the shuffle: group by key, then sort keys
for k, v in mapped:
    groups.setdefault(k, []).append(v)
print("shuffled   :", dict(sorted(groups.items())))

for item in sorted(groups):
    print("reduce     :", reduce_fn(item, groups[item]))

map output : [('samosa', 18), ('vada pav', 20), ('poha', 25), ('samosa', 18)] ... shuffled : {'chai': [15, 15, 15], 'poha': [25, 25], 'samosa': [18, 18], 'vada pav': [20]} reduce : ('chai', 3, 45) reduce : ('poha', 2, 50) reduce : ('samosa', 2, 36) reduce : ('vada pav', 1, 20)

if f[0] == "order_id":The header line produces no pair at all: that is why Map input records (300,001) was one more than Map output records (300,000).
groups = {} # the shuffle: group by key, then sort keysIn Hadoop this is the shuffle and sort: partition, copy across the network, merge, sort by key.
return (item, len(amounts), sum(amounts))The reducer sees every value of one key together, so counting and summing is trivial.

Run it on your PC: the Module 8 lab and your first job

Module 8 needs YARN, so it has its own lab cluster: a NameNode, a ResourceManager, a JobHistory Server and three workers that each run a DataNode and a NodeManager. Stop the Module 5 lab first (docker compose -p lab5 stop); together they need more memory than a laptop has. Put docker-compose.yaml and config in a folder called mr-lab and start it. Containers are small (384 MB per task) so everything fits in about 5 GB of RAM.

Start the lab

# Lab 8: HDFS + YARN. Each worker runs a DataNode AND a NodeManager (so maps can be data-local).
x-hadoop: &hadoop
  image: apache/hadoop:3.5.0
  env_file: [./config]
services:
  namenode:
    <<: *hadoop
    hostname: namenode
    command: ["hdfs", "namenode"]
    environment: { ENSURE_NAMENODE_DIR: "/tmp/hadoop-hadoop/dfs/name" }
  resourcemanager:
    <<: *hadoop
    hostname: resourcemanager
    command: ["yarn", "resourcemanager"]
  historyserver:
    <<: *hadoop
    hostname: historyserver
    command: ["mapred", "historyserver"]
  worker1:
    <<: *hadoop
    hostname: worker1
    command: ["bash", "-c", "hdfs --daemon start datanode && yarn nodemanager"]
    depends_on: [namenode, resourcemanager]
  worker2:
    <<: *hadoop
    hostname: worker2
    command: ["bash", "-c", "hdfs --daemon start datanode && yarn nodemanager"]
    depends_on: [namenode, resourcemanager]
  worker3:
    <<: *hadoop
    hostname: worker3
    command: ["bash", "-c", "hdfs --daemon start datanode && yarn nodemanager"]
    depends_on: [namenode, resourcemanager]

# ---- config (same folder) ----
# CORE-SITE.XML_fs.defaultFS=hdfs://namenode
# HDFS-SITE.XML_dfs.namenode.rpc-address=namenode:8020
# HDFS-SITE.XML_dfs.replication=3
# YARN-SITE.XML_yarn.resourcemanager.hostname=resourcemanager
# YARN-SITE.XML_yarn.nodemanager.aux-services=mapreduce_shuffle
# YARN-SITE.XML_yarn.nodemanager.resource.memory-mb=2048
# YARN-SITE.XML_yarn.nodemanager.resource.cpu-vcores=2
# YARN-SITE.XML_yarn.scheduler.minimum-allocation-mb=256
# YARN-SITE.XML_yarn.nodemanager.pmem-check-enabled=false
# YARN-SITE.XML_yarn.nodemanager.vmem-check-enabled=false
# YARN-SITE.XML_yarn.nodemanager.disk-health-checker.max-disk-utilization-per-disk-percentage=99.5
# YARN-SITE.XML_yarn.log-aggregation-enable=true
# YARN-SITE.XML_yarn.nodemanager.delete.debug-delay-sec=600
# MAPRED-SITE.XML_mapreduce.framework.name=yarn
# MAPRED-SITE.XML_yarn.app.mapreduce.am.env=HADOOP_MAPRED_HOME=/opt/hadoop
# MAPRED-SITE.XML_mapreduce.map.env=HADOOP_MAPRED_HOME=/opt/hadoop
# MAPRED-SITE.XML_mapreduce.reduce.env=HADOOP_MAPRED_HOME=/opt/hadoop
# MAPRED-SITE.XML_yarn.app.mapreduce.am.resource.mb=512
# MAPRED-SITE.XML_yarn.app.mapreduce.am.command-opts=-Xmx400m
# MAPRED-SITE.XML_mapreduce.map.memory.mb=384
# MAPRED-SITE.XML_mapreduce.map.java.opts=-Xmx300m
# MAPRED-SITE.XML_mapreduce.reduce.memory.mb=384
# MAPRED-SITE.XML_mapreduce.reduce.java.opts=-Xmx300m
# MAPRED-SITE.XML_mapreduce.task.io.sort.mb=64
# MAPRED-SITE.XML_mapreduce.jobhistory.address=historyserver:10020
# MAPRED-SITE.XML_mapreduce.jobhistory.webapp.address=historyserver:19888

# Run, in the mr-lab folder:
#   docker compose -p lab8 up -d
#   docker exec lab8-resourcemanager-1 yarn node -list

$ docker compose -p lab8 up -d Container lab8-worker1-1 Started $ docker exec lab8-resourcemanager-1 yarn node -list Total Nodes:3 Node-Id Node-State Node-Http-Address Number-of-Running-Containers worker1:35119 RUNNING worker1:8042 0 worker2:46619 RUNNING worker2:8042 0 worker3:45331 RUNNING worker3:8042 0 $ docker exec lab8-namenode-1 hdfs dfsadmin -report | grep -E "^Name|Live" Live datanodes (3): Name: 172.22.0.5:9866 (lab8-worker2-1.lab8_default) Name: 172.22.0.6:9866 (lab8-worker3-1.lab8_default) Name: 172.22.0.7:9866 (lab8-worker1-1.lab8_default)

Real output. Three NodeManagers and three DataNodes, one of each per worker. The ResourceManager web UI is on port 8088 inside the network, the JobHistory Server on 19888.

Item revenue as a MapReduce job

// ItemRevenue.java: orders per item and revenue per item, as a MapReduce job (new API).
// Input lines: order_id,item,amount      Output: item <TAB> orders <TAB> revenue
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class ItemRevenue {

  public static class OrderMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final Text item = new Text();
    private final IntWritable amount = new IntWritable();

    @Override
    protected void map(LongWritable offset, Text line, Context ctx) throws IOException, InterruptedException {
      String[] f = line.toString().split(",");
      if (f.length != 3 || f[0].equals("order_id")) return;     // skip header / broken lines
      item.set(f[1]);
      amount.set(Integer.parseInt(f[2]));
      ctx.write(item, amount);                                  // ("samosa", 18)
    }
  }

  public static class SumReducer extends Reducer<Text, IntWritable, Text, Text> {
    @Override
    protected void reduce(Text item, Iterable<IntWritable> amounts, Context ctx) throws IOException, InterruptedException {
      long orders = 0, revenue = 0;
      for (IntWritable a : amounts) { orders++; revenue += a.get(); }
      ctx.write(item, new Text(orders + "\t" + revenue));      // samosa  50012  900216
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();   // accepts -D key=value
    Job job = Job.getInstance(conf, "item revenue");
    job.setJarByClass(ItemRevenue.class);
    job.setMapperClass(OrderMapper.class);
    job.setReducerClass(SumReducer.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(IntWritable.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    FileInputFormat.addInputPath(job, new Path(rest[0]));
    FileOutputFormat.setOutputPath(job, new Path(rest[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

// In the mr-lab folder (make_canteen.py is from Lesson 4.3):
//   docker exec lab8-namenode-1 mkdir -p /tmp/mr
//   docker cp ItemRevenue.java lab8-namenode-1:/tmp/mr/
//   docker cp canteen_orders.csv lab8-namenode-1:/tmp/mr/
//   docker exec -it lab8-namenode-1 bash
// Inside the container:
//   cd /tmp/mr
//   hdfs dfs -mkdir -p /user/hadoop/data
//   hdfs dfs -put canteen_orders.csv /user/hadoop/data/
//   mkdir -p classes && javac -cp $(hadoop classpath) -d classes *.java && jar cf mr.jar -C classes .
//   hadoop jar mr.jar ItemRevenue /user/hadoop/data/canteen_orders.csv /out/item_revenue
//   hdfs dfs -cat /out/item_revenue/part-r-00000

2026-10-21 16:52:22 INFO DefaultNoHARMFailoverProxyProvider:64 - Connecting to ResourceManager at resourcemanager/172.22.0.2:8032 2026-10-21 16:52:22 INFO JobResourceUploader:907 - Disabling Erasure Coding for path: /tmp/hadoop-yarn/staging/hadoop/.staging/job_1792601400460_0001 2026-10-21 16:52:22 INFO FileInputFormat:303 - Total input files to process : 1 2026-10-21 16:52:23 INFO JobSubmitter:203 - number of splits:1 2026-10-21 16:52:23 INFO JobSubmitter:299 - Submitting tokens for job: job_1792601400460_0001 2026-10-21 16:52:23 INFO JobSubmitter:300 - Executing with tokens: [] 2026-10-21 16:52:23 INFO Configuration:2885 - resource-types.xml not found 2026-10-21 16:52:23 INFO ResourceUtils:476 - Unable to find 'resource-types.xml'. 2026-10-21 16:52:24 INFO YarnClientImpl:356 - Submitted application application_1792601400460_0001 2026-10-21 16:52:24 INFO Job:1681 - The url to track the job: http://resourcemanager:8088/proxy/application_1792601400460_0001/ 2026-10-21 16:52:24 INFO Job:1726 - Running job: job_1792601400460_0001 2026-10-21 16:52:32 INFO Job:1747 - Job job_1792601400460_0001 running in uber mode : false 2026-10-21 16:52:32 INFO Job:1754 - map 0% reduce 0% 2026-10-21 16:52:38 INFO Job:1754 - map 100% reduce 0% 2026-10-21 16:52:44 INFO Job:1754 - map 100% reduce 100% 2026-10-21 16:52:45 INFO Job:1765 - Job job_1792601400460_0001 completed successfully 2026-10-21 16:52:45 INFO Job:1772 - Counters: 54 File System Counters FILE: Number of bytes read=3599992 FILE: Number of bytes written=7836231 FILE: Number of read operations=0 FILE: Number of large read operations=0 FILE: Number of write operations=0 HDFS: Number of bytes read=4692125 HDFS: Number of bytes written=117 HDFS: Number of read operations=8 HDFS: Number of large read operations=0 HDFS: Number of write operations=2 HDFS: Number of bytes read erasure-coded=0 Job Counters Launched map tasks=1 Launched reduce tasks=1 Data-local map tasks=1 Total time spent by all maps in occupied slots (ms)=7876 Total time spent by all reduces in occupied slots (ms)=6122 Total time spent by all map tasks (ms)=3938 Total time spent by all reduce tasks (ms)=3061 Total vcore-milliseconds taken by all map tasks=3938 Total vcore-milliseconds taken by all reduce tasks=3061 Total megabyte-milliseconds taken by all map tasks=2016256 Total megabyte-milliseconds taken by all reduce tasks=1567232 Map-Reduce Framework Map input records=300001 Map output records=300000 Map output bytes=2999986 Map output materialized bytes=3599992 Input split bytes=116 Combine input records=0 Combine output records=0 Reduce input groups=6 Reduce shuffle bytes=3599992 Reduce input records=300000 Reduce output records=6 Spilled Records=600000 Shuffled Maps =1 Failed Shuffles=0 Merged Map outputs=1 GC time elapsed (ms)=83 CPU time spent (ms)=3240 Physical memory (bytes) snapshot=494002176 Virtual memory (bytes) snapshot=4358275072 Total committed heap usage (bytes)=266338304 Peak Map Physical memory (bytes)=290557952 Peak Map Virtual memory (bytes)=2176516096 Peak Reduce Physical memory (bytes)=203444224 Peak Reduce Virtual memory (bytes)=2181758976 Shuffle Errors BAD_ID=0 CONNECTION=0 IO_ERROR=0 WRONG_LENGTH=0 WRONG_MAP=0 WRONG_REDUCE=0 File Input Format Counters Bytes Read=4692009 File Output Format Counters Bytes Written=117 real 0m25.334s user 0m4.889s sys 0m0.415s Found 2 items -rw-r--r-- 3 hadoop supergroup 0 2026-10-21 16:52 /out/item_revenue/_SUCCESS -rw-r--r-- 3 hadoop supergroup 117 2026-10-21 16:52 /out/item_revenue/part-r-00000 chai 50170 752550 dosa 49948 2247660 idli 49984 1499520 poha 49891 1247275 samosa 50021 900378 vada pav 49986 999720

Real output, unedited. 54 counters in total; the ones that matter most today are Launched map tasks, Data-local map tasks, Map input/output records and Reduce input groups.

90 times the data, and where each task ran

# attempts.sh JOB_ID : every task attempt of a finished job (node, time, state), from the JobHistory REST API
J=$1; ID=${J#job_}
for kind in m r; do
  for t in $(curl -s http://historyserver:19888/ws/v1/history/mapreduce/jobs/$J/tasks | grep -o "task_${ID}_${kind}_[0-9]*" | sort -u); do
    curl -s http://historyserver:19888/ws/v1/history/mapreduce/jobs/$J/tasks/$t/attempts | python3 -c '
import sys, json
t = json.load(sys.stdin)["taskAttempts"]["taskAttempt"]
for a in (t if isinstance(t, list) else [t]):
    node = a["nodeHttpAddress"].split(":")[0]
    print("%-40s %-9s %-8s %5.1f s  %s" % (a["id"], a["state"], node, a["elapsedTime"] / 1000, a.get("diagnostics", "")[:70]))'
  done
done

# Inside the namenode container, in /tmp/mr:
#   for i in $(seq 30); do tail -n +2 canteen_orders.csv; done > orders_x30.csv
#   cat orders_x30.csv orders_x30.csv orders_x30.csv > orders_x90.csv
#   hdfs dfs -put orders_x90.csv /user/hadoop/data/
#   hadoop jar mr.jar ItemRevenue /user/hadoop/data/orders_x90.csv /out/item_revenue_x90
#   bash attempts.sh job_<your job id>

$ hadoop jar mr.jar ItemRevenue /user/hadoop/data/orders_x90.csv /out/item_revenue_x90 JobSubmitter:203 - number of splits:4 YarnClientImpl:356 - Submitted application application_1792601400460_0003 Job:1754 - map 0% reduce 0% Job:1754 - map 25% reduce 0% Job:1754 - map 60% reduce 0% Job:1754 - map 84% reduce 0% Job:1754 - map 100% reduce 0% Job:1754 - map 100% reduce 78% Job:1754 - map 100% reduce 100% Job:1765 - Job job_1792601400460_0003 completed successfully Killed map tasks=1 Launched map tasks=5 Launched reduce tasks=1 Data-local map tasks=5 Total time spent by all map tasks (ms)=144021 Map input records=27000000 Reduce input groups=6 real 1m8.106s $ hdfs dfs -cat /out/item_revenue_x90/part-r-00000 chai 4515300 67729500 dosa 4495320 202289400 idli 4498560 134956800 poha 4490190 112254750 samosa 4501890 81034020 vada pav 4498740 89974800 $ bash attempts.sh job_1792601400460_0003 attempt_1792601400460_0003_m_000000_0 SUCCEEDED worker2 43.9 s attempt_1792601400460_0003_m_000000_1 KILLED worker3 11.7 s Speculation: attempt_1792601400460_0003_m_000000_0 succeeded first! attempt_1792601400460_0003_m_000001_0 SUCCEEDED worker3 30.6 s attempt_1792601400460_0003_m_000002_0 SUCCEEDED worker1 29.1 s attempt_1792601400460_0003_m_000003_0 SUCCEEDED worker2 28.6 s attempt_1792601400460_0003_r_000000_0 SUCCEEDED worker1 25.1 s

Real output (the job log is shortened to the progress and key counters). Your speculative attempt may or may not appear: it depends on whether one map runs clearly slower than the others. The built-in “mapred job -history all” printer crashed on our 7-character host names (worker1), so attempts.sh asks the JobHistory Server REST API instead.

8 Line by line

One record's trip through map, shuffle and reduce

Follow how four lines become two output rows. Watch the groups dictionary: it is the shuffle.

Worked examples: real problems, solved and traced in memory

Three MapReduce design and sizing problems: choosing the key, estimating a job, and a reducer that must be careful.

Example 1BeginnerDesign · key

Design a job: UPI transactions per bank

Problem statement

Input lines look like txn_id,payer_bank,payee_bank,amount. The bank wants the number of transactions and the total amount sent by each payer bank. What should map emit, and what does reduce do?

Input

4 sample lines

Expected output

HDFC 1 1200
SBI 3 800

Approach

The answer is per payer bank, so the key is payer_bank.
The value is the amount; the reducer counts and sums.
Simulate it on four lines.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

Example 2IntermediateSizing · tasks

How many map tasks, and how long?

Problem statement

A 1 TB file in HDFS with 128 MB blocks. The cluster has 20 workers with 8 map containers each. A map task takes 40 s. How many map tasks run, in how many waves, and roughly how long is the map phase?

Input

1 TB · 128 MB splits · 160 containers · 40 s/task

Expected output

8192 map tasks, 52 waves, about 35 minutes

Approach

Map tasks = number of splits ≈ file size / block size.
Waves = tasks / containers, rounded up.
Map phase ≈ waves × task time.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

Example 3AdvancedCorrectness · averages

Why you can't average the averages

Problem statement

Two map tasks each compute the average dosa price they saw and send only that average to the reducer, which averages the two. Map 1 saw 3 orders at ₹45; map 2 saw 1 order at ₹30 (a discount). Is the reducer's answer right? How should the map output look instead?

Input

map 1: 45, 45, 45 · map 2: 30

Expected output

true 41.25 · average of averages 37.50 · from (sum, count) 41.25

Approach

Compute the true average over all four orders.
Compute the average of the two map averages.
Fix: emit (sum, count) and divide only in the reducer.

Solution: run it live

What happens in memory, step by step

Recorded from a real Python run: press Next to execute one line at a time. Watch variables change in each frame, and lists and dictionaries (numbered #1, #2 …) fill up as objects.

9 Quiz and revision

Check yourself

Quick revision

Lesson 8.1 in one screen

map(k1, v1) → list(k2, v2); shuffle and sort by key; reduce(k2, list(v2)) → list(k3, v3).
You write Mapper, Reducer and a driver; Hadoop splits input, schedules tasks near data, shuffles, retries and writes output.
One map task per input split; reducers = mapreduce.job.reduces (default 1); output files part-r-NNNNN + _SUCCESS.
Lab: 4.7 MB → 1 map, 25 s (Python: 0.1 s); 422 MB → 4 maps on 3 workers, all data-local, 68 s; one speculative attempt killed.
Map output goes to local disk, not HDFS; the output folder must not exist before the job.
Design question: what is the key? Only values with the same key ever meet.
Good fit: big batch scans and aggregations. Poor fit: iterative and interactive work (use Spark, Tez).

Interview and viva questions

What is MapReduce in Hadoop?

MapReduce is Hadoop's programming model and framework for batch processing. You write a map function that turns each input record into key-value pairs and a reduce function that combines all values of each key. The framework splits the input, runs map tasks in parallel on the nodes that store the data, shuffles and sorts the intermediate pairs by key, runs reduce tasks and writes the result to HDFS, retrying failed tasks automatically.

What are the phases of a MapReduce job?

Input splitting and reading (InputFormat and RecordReader), the map phase, the shuffle and sort (partitioning, spilling, copying, merging), the reduce phase and output writing (OutputFormat). An optional combiner can run on map output before the shuffle.

Is MapReduce still used?

Classic Java MapReduce is rarely chosen for new projects; Spark, Hive on Tez, Flink and cloud warehouses are more common. It still runs in many established Hadoop clusters (including tools like DistCp and Hadoop Archive, which are MapReduce jobs), and its model of map, shuffle by key and reduce underlies all of those newer engines, so it remains a core interview and design topic.

Why is MapReduce slow for small data?

Every job starts an ApplicationMaster and task JVMs in YARN containers, sorts and writes map output to local disk and copies it over the network. That fixed overhead is tens of seconds; in our lab a 4.7 MB file took 25 s against 0.1 s in plain Python. The model pays off when the data is too big for one machine to read in time.

What is speculative execution in MapReduce?

When a task runs noticeably slower than others of the same job, the MRAppMaster can start a duplicate attempt on another node. Whichever attempt finishes first is used and the other is killed. It protects jobs from slow nodes; it is controlled by mapreduce.map.speculative and mapreduce.reduce.speculative (both true by default).

Practice problems

MapReduce thinking exercises, using the real numbers from our lab.

P1Easy

Records lost or not?

Map input records=300001 and Map output records=300000. Explain the difference with one line of code.

Input

two counters

Expected output

1 line produced no pair: the header 'order_id,item,amount'

Show solution and explanation

print(300001 - 300000, "line produced no pair: the header 'order_id,item,amount'")

1 line produced no pair: the header 'order_id,item,amount'

How it works: The mapper returns early for the header, so it emits nothing for that record.

P2Easy

Check the reducer

Using the output, compute the price of one dosa and one chai.

Input

dosa 49948 2247660 · chai 50170 752550

Expected output

dosa 45.0 chai 15.0

Show solution and explanation

print("dosa", 2247660 / 49948, "chai", 752550 / 50170)

dosa 45.0 chai 15.0

How it works: 45.0 and 15.0: the menu prices, so the sums are consistent.

P3Easy

Scale check

orders_x90 has 90 copies of 3 lakh orders. Check that chai's count in the big job equals 90 × the small job.

Input

50,170 and 4,515,300

Expected output

True

Show solution and explanation

print(50170 * 90 == 4515300)

True

How it works: True. The same code gives exactly the same answer at 90 times the size.

P4Medium

Word count by hand

Write map and reduce for word count over three lines, with a dictionary as the shuffle.

Input

3 lines of feedback

Expected output

chai 2
cold 1
dosa 1
hot 1
late 1
the 2
was 3

Show solution and explanation

lines = ["the chai was hot", "the dosa was cold", "chai was late"]
groups = {}
for line in lines:
    for word in line.split():
        groups.setdefault(word, []).append(1)
for w in sorted(groups):
    print(w, sum(groups[w]))

chai 2 cold 1 dosa 1 hot 1 late 1 the 2 was 3

How it works: Every word becomes (word, 1); the reducer sums the ones. Lesson 8.6 runs this on 2 lakh lines.

P5Medium

Inverted index

For each word, list the line numbers it appears in.

Input

3 lines

Expected output

chai [1, 3]
cold [2, 3]
dosa [2]
hot [1]

Show solution and explanation

lines = ["chai hot", "dosa cold", "chai cold"]
index = {}
for n, line in enumerate(lines, 1):
    for w in set(line.split()):
        index.setdefault(w, []).append(n)
for w in sorted(index):
    print(w, index[w])

chai [1, 3] cold [2, 3] dosa [2] hot [1]

How it works: Map emits (word, line number); reduce collects the list. This is how search indexes are built.

P6Medium

Waves of tasks

Our lab has 3 workers with 2048 MB each; a map container is 384 MB and the MRAppMaster takes 512 MB on one worker. How many maps can run at once?

Input

3 × 2048 MB · 384 MB · AM 512 MB

Expected output

13 maps at once (about)

Show solution and explanation

per_node = 2048 // 384
print(per_node * 3 - (512 + 383) // 384, "maps at once (about)")

13 maps at once (about)

How it works: About 13: 5 per worker, minus room for the AM. Our 4 maps all ran in one wave.

P7Business problem

Rent a cluster or wait?

A monthly report scans 50 TB. One server reads at 300 MB/s. A 100-node cluster reads in parallel at the same speed per node, costing ₹40 per node-hour. Compare times and cluster cost.

Input

50 TB · 300 MB/s · 100 nodes · ₹40/node-hour

Expected output

one server 49 h, cluster 0.5 h, cluster cost Rs 4,000

Show solution and explanation

one = 50 * 1024 * 1024 / 300 / 3600
many = one / 100
print(f"one server {one:.0f} h, cluster {many:.1f} h, cluster cost Rs {100 * max(1, many) * 40:,.0f}")

one server 49 h, cluster 0.5 h, cluster cost Rs 4,000

How it works: About 49 hours vs half an hour for ₹4,000 (illustrative), and the report is ready the same morning.

P8Advanced

Top-N per mapper

Each map task keeps only its local top 2 items by revenue; the reducer merges them into the global top 2. Compare with the true top 2 for these two maps.

Input

2 maps, partial revenue per item

Expected output

true top 2: [('dosa', 1750), ('idli', 700)]
from local top 2s: [('dosa', 1750), ('poha', 700)]

Show solution and explanation

m1 = {"dosa": 900, "idli": 600, "chai": 300}
m2 = {"dosa": 850, "poha": 700, "idli": 100}
total = {}
for m in (m1, m2):
    for k, v in m.items():
        total[k] = total.get(k, 0) + v
print("true top 2:", sorted(total.items(), key=lambda x: -x[1])[:2])
local = [sorted(m.items(), key=lambda x: -x[1])[:2] for m in (m1, m2)]
merged = {}
for top in local:
    for k, v in top:
        merged[k] = merged.get(k, 0) + v
print("from local top 2s:", sorted(merged.items(), key=lambda x: -x[1])[:2])

true top 2: [('dosa', 1750), ('idli', 700)] from local top 2s: [('dosa', 1750), ('poha', 700)]

How it works: Wrong answer: map 2 dropped idli's 100, so idli shows 600 instead of 700 and poha takes second place. Local top-N is only safe when each key's full total is known in one place, e.g. after a first job that sums per item; then a second step picks the top N.

Real-world scenario questions (asked in company interviews)

MapReduce basics appear in every Hadoop interview, usually as “explain with an example” and “design a job for X”.

Concepts · basicsFresher

“Explain MapReduce with word count”

The most common opening question.

The interviewer is checking: the three phases and who does what.

Show a strong answer

Map: for each line, emit (word, 1). Runs in parallel, one task per split, near the data.
Shuffle and sort (framework): all pairs with the same word go to the same reducer, sorted by key.
Reduce: sum the ones for each word and write (word, count) to HDFS, one part file per reducer.

Design · keysFresher

Design a job: average order value per city per month

Input: order_id, city, date, amount.

The interviewer is checking: composite keys and correct averaging.

Show a strong answer

Key = (city, yyyy-mm); value = amount (or (amount, 1) if a combiner will be used).
Reducer: sum and count, then divide once. Never average averages.
One output line per city-month; the number of reducers decides the number of output files.

Operations · performanceExperienced

“Our MapReduce job on 2 GB takes 3 minutes. Python on a laptop takes 20 seconds.”

A manager asks why Hadoop is “slow”.

The interviewer is checking: fixed overhead vs scale.

Show a strong answer

Each job pays fixed costs: AM and container start-up, JVM launch, sort and spill to disk, shuffle over the network.
For 2 GB that overhead dominates; for 20 TB the parallel read across hundreds of nodes dominates and wins by far.
Options: don't use a cluster for small data, use uber mode for tiny jobs, or move to Spark/Hive on Tez for interactive work.

Internals · failuresExperienced

“The job counters show Killed map tasks=1 but the job succeeded. Is something wrong?”

A junior engineer is worried after reading the counters.

The interviewer is checking: speculation vs failure.

Show a strong answer

Killed is not failed: with speculative execution the slower of two attempts is killed once the other succeeds.
Check the attempt diagnostics (“Speculation: … succeeded first!”) in the JobHistory UI or REST API.
Failed task attempts (non-zero exit codes) would show up as “Failed map tasks” and are retried up to 4 times by default.

Thinking in MapReduce

Counting votes the way India does

MapReduce

Words you'll meet in this lesson

From our lab (Hadoop 3.5.0, HDFS + YARN, 3 workers)

MapReduce at three levels

Beginner Counting day

Intermediate What you write, what Hadoop does

Advanced Thinking in keys

Common MapReduce patterns

The pieces of a MapReduce job on YARN

Client

MRAppMaster

Map and reduce tasks

JobHistory Server

Our first MapReduce job, on a real cluster

Write the map and the reduce

Submit the job

Read the counters

Look at the output

Feed it 90 times more data

Be honest about speed

One job across three workers

Try it yourself: a job in the playground

MapReduce at work

1. Building a search index

2. Telecom and bank batch jobs

3. Log analytics

4. Under SQL engines

Simulate map, shuffle and reduce

Run it on your PC: the Module 8 lab and your first job

Start the lab

Item revenue as a MapReduce job

90 times the data, and where each task ran

One record's trip through map, shuffle and reduce

Worked examples: real problems, solved and traced in memory

Design a job: UPI transactions per bank

How many map tasks, and how long?

Why you can't average the averages

Check yourself

Quick revision

Lesson 8.1 in one screen

Interview and viva questions

Practice problems

Records lost or not?

Check the reducer

Scale check

Word count by hand

Inverted index

Waves of tasks

Rent a cluster or wait?

Top-N per mapper

Real-world scenario questions (asked in company interviews)

“Explain MapReduce with word count”

Design a job: average order value per city per month

“Our MapReduce job on 2 GB takes 3 minutes. Python on a laptop takes 20 seconds.”

“The job counters show Killed map tasks=1 but the job succeeded. Is something wrong?”