Word count end to end

Two lakh complaints and compliments

A MapReduce job, end to end

Words you'll meet in this lesson

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

The whole job at three levels

Two Java APIs, many ways to run

Hadoop ships two MapReduce Java APIs. Old books and Stack Overflow answers mix them, which is a classic source of compile errors.

Ways to launch a word count

From an empty file to a result in HDFS

Run on the Module 8 lab cluster (Lesson 8.1) with wordcount_lab.sh and kill_task.sh from “Run it on your PC”.

1. Write the mapper and reducer

   // Mapper<input key, input value, output key, output value>
   public void map(LongWritable offset, Text line, Context context) … {
     StringTokenizer it = new StringTokenizer(line.toString().toLowerCase(), " \t.,!?;:\"'()");
     while (it.hasMoreTokens()) { word.set(it.nextToken()); context.write(word, ONE); }
   }
   // Reducer<Text, IntWritable, Text, IntWritable>
   public void reduce(Text word, Iterable<IntWritable> ones, Context context) … {
     int sum = 0;
     for (IntWritable one : ones) sum += one.get();
     result.set(sum); context.write(word, result);
   }

   The four generic types must line up: the mapper's output types (Text, IntWritable) are the reducer's input types. word and result are reused objects: Hadoop serializes them on every write, so reuse is safe and saves millions of allocations.

2. Write the driver

   String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
   Job job = Job.getInstance(conf, "word count");
   job.setJarByClass(WordCount.class);
   job.setMapperClass(TokenizerMapper.class);
   job.setReducerClass(IntSumReducer.class);
   job.setOutputKeyClass(Text.class);
   job.setOutputValueClass(IntWritable.class);
   FileInputFormat.addInputPath(job, new Path(rest[0]));
   FileOutputFormat.setOutputPath(job, new Path(rest[1]));
   System.exit(job.waitForCompletion(true) ? 0 : 1);

   setJarByClass tells Hadoop which jar to ship to the workers. Input format defaults to TextInputFormat (key = byte offset, value = line), output to TextOutputFormat (key TAB value).

3. Compile and package

   javac -cp $(hadoop classpath) -d wc_classes WordCount.java
   jar cf wc.jar -C wc_classes .
   jar tf wc.jar | grep class

   WordCount$IntSumReducer.class WordCount$TokenizerMapper.class WordCount.class
   Nested classes compile to separate files with a $ in the name. They must be static: Hadoop creates them by reflection with no outer object.

4. Submit and watch

   hadoop jar wc.jar WordCount /user/hadoop/data/feedback.txt /out/wc

   JobSubmitter:203 - number of splits:1 YarnClientImpl:356 - Submitted application application_1792601400460_0030 Job:1681 - The url to track the job: http://resourcemanager:8088/proxy/application_1792601400460_0030/ Job:1747 - Job job_1792601400460_0030 running in uber mode : false Job:1754 - map 0% reduce 0% Job:1754 - map 100% reduce 0% Job:1754 - map 100% reduce 100% Job:1765 - Job job_1792601400460_0030 completed successfully Map input records=200000 Map output records=1033238 Reduce input groups=31 Reduce output records=31
   One split (5 MB fits in one block). The tracking URL points at the ResourceManager UI on port 8088; after the job ends the same details live on the JobHistory Server (port 19888). 1,033,238 words became 1,033,238 pairs, all shuffled to one reducer, which found 31 different words.

5. Read the result

   hdfs dfs -cat /out/wc/part-r-00000 | head -5
   hdfs dfs -cat /out/wc/part-r-00000 | sort -k2,2nr | head -8

   amazing 17687 chai 33431 cold 8452 costly 8410 crispy 17585 was 200000 the 166921 loved 33570 dosa 33566 chai 33431 samosa 33431 improve 33398 please 33398
   The file is sorted by word (the shuffle sorts keys), not by count; sorting by count is a second step. “was” appears once in every line: 200,000. Good words (tasty, crispy…) appear about twice as often as bad ones (cold, stale…).

6. Make it fail, twice, before it starts

   hadoop jar wc.jar WordCount /user/hadoop/data/feedback.txt /out/wc
   hadoop jar wc.jar WordCount /user/hadoop/data/feedbak.txt /out/wc2

   Exception in thread "main" org.apache.hadoop.mapred.FileAlreadyExistsException: Output directory hdfs://namenode/out/wc already exists Exception in thread "main" org.apache.hadoop.mapreduce.lib.input.InvalidInputException: Input path does not exist: hdfs://namenode/user/hadoop/data/feedbak.txt
   Both are caught by the client before anything is sent to YARN: the output check and the split calculation happen at submission. Cheap, early failures are a feature.

7. Kill a running map task

   docker exec lab8-worker2-1 kill -9 <pid of the map task's YarnChild JVM>
   bash attempts.sh job_1792601400460_0033

   Task Id : attempt_1792601400460_0033_m_000000_0, Status : FAILED Container exited with a non-zero exit code 137. Job job_1792601400460_0033 completed successfully Failed map tasks=1 Launched map tasks=5 attempt_…_m_000000_0 FAILED worker2 7.9 s attempt_…_m_000000_1 SUCCEEDED worker3 21.7 s attempt_…_m_000000_2 KILLED worker2 4.4 s Speculation: attempt_…_m_000000_1 succeeded first! chai 668620 was 4000000
   The AM saw the container die (137 = 128 + signal 9) and launched attempt _1 on worker3. It also started a speculative _2, which lost the race and was killed. The answer is exactly 20 × the small one (chai 20 × 33431 = 668620): a failed attempt's partial output is thrown away, never mixed in.

The job's life, including a crash

Six steps of the 20× run: submit, ApplicationMaster, maps, a killed task, the retry and the reduce.

Try it yourself: a job that survives a failure

Upload a file, run the word count, then fail a node and run it again. Watch which tasks are rerun.

Word count's real cousins

Simplified; organisations differ in tools and policies.

Word count, exactly like the Java version

The tokenizer below uses the same delimiters as WordCount.java (space, tab and .,!?;:"'(), after lowercasing), so for the same lines it gives the same counts as the real job.

DELIMS = " \t.,!?;:\"'()"

def tokens(line):                         # like java.util.StringTokenizer(line.toLowerCase(), DELIMS)
    word, out = "", []
    for ch in line.lower():
        if ch in DELIMS:
            if word: out.append(word)
            word = ""
        else:
            word += ch
    if word: out.append(word)
    return out

lines = ["the idli was fresh", "Why is the chai was cold?", "loved the vada pav was tasty!", "the chai was hot"]
pairs = [(w, 1) for line in lines for w in tokens(line)]       # map
print(len(pairs), "pairs, e.g.", pairs[:5])

counts = {}
for w, one in pairs:                                           # shuffle + reduce
    counts[w] = counts.get(w, 0) + one
for w in sorted(counts):
    print(w, counts[w])

• for ch in line.lower():toLowerCase() first, so “Why” and “why” are one key.
• if ch in DELIMS:Punctuation splits words just like spaces: “cold?” becomes “cold”.
• pairs = [(w, 1) for line in lines for w in tokens(line)] # mapOne pair per word: our 2 lakh lines produced 1,033,238 pairs.

Run it on your PC: build, run and break word count

Use the Module 8 lab from Lesson 8.1. Create feedback.txt with make_feedback.py, copy it and WordCount.java into the namenode container, put the text into HDFS, then run wordcount_lab.sh inside the container. kill_task.sh runs on your PC because it needs docker exec into a worker.

// WordCount.java: the classic first MapReduce program (new org.apache.hadoop.mapreduce API).
import java.io.IOException;
import java.util.StringTokenizer;
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 WordCount {

  public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private static final IntWritable ONE = new IntWritable(1);
    private final Text word = new Text();

    @Override
    public void map(LongWritable offset, Text line, Context context) throws IOException, InterruptedException {
      StringTokenizer it = new StringTokenizer(line.toString().toLowerCase(), " \t.,!?;:\"'()");
      while (it.hasMoreTokens()) {
        word.set(it.nextToken());
        context.write(word, ONE);                 // ("chai", 1)
      }
    }
  }

  public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private final IntWritable result = new IntWritable();

    @Override
    public void reduce(Text word, Iterable<IntWritable> ones, Context context) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable one : ones) sum += one.get();
      result.set(sum);
      context.write(word, result);                // ("chai", 41873)
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
    if (rest.length != 2) { System.err.println("Usage: wordcount <in> <out>"); System.exit(2); }
    Job job = Job.getInstance(conf, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job, new Path(rest[0]));
    FileOutputFormat.setOutputPath(job, new Path(rest[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

// ---- make_feedback.py (run on your PC) ----
// # make_feedback.py: create feedback.txt, 2 lakh short canteen feedback lines (same file every time)
// import random
// random.seed(11)
// items = ["chai", "samosa", "vada pav", "poha", "idli", "dosa"]
// good = ["tasty", "hot", "fresh", "crispy", "perfect", "good", "nice", "amazing"]
// bad = ["cold", "late", "oily", "salty", "costly", "stale", "slow"]
// starts = ["the", "today the", "my", "loved the", "please improve the", "why is the"]
// with open("feedback.txt", "w") as f:
//     for _ in range(200_000):
//         item = random.choice(items)
//         words = random.choice(good if random.random() < 0.7 else bad)
//         f.write(f"{random.choice(starts)} {item} was {words}\n")
// print(open("feedback.txt").read().count("\n"), "lines")

// ---- wordcount_lab.sh (inside the namenode container) ----
// # wordcount_lab.sh: compile, package and run WordCount on YARN; read the result; make it fail on purpose
// cd /tmp/mr
// echo "== 1. compile and package"
// mkdir -p wc_classes
// javac -cp $(hadoop classpath) -d wc_classes WordCount.java && jar cf wc.jar -C wc_classes .
// jar tf wc.jar | grep class
// echo "== 2. run on feedback.txt"
// hdfs dfs -rm -r -f -skipTrash /out/wc > /dev/null 2>&1
// hadoop jar wc.jar WordCount /user/hadoop/data/feedback.txt /out/wc 2>&1 \
//   | grep -vE "resource-types|ResourceUtils|Executing with tokens|Disabling Erasure" | sed -E 's/^[0-9-]+ [0-9:]+ INFO  //'
// echo "== 3. the result"
// hdfs dfs -ls /out/wc | awk '{print $5, $8}'
// hdfs dfs -cat /out/wc/part-r-00000 | head -5
// echo "-- top 8 words"
// hdfs dfs -cat /out/wc/part-r-00000 | sort -k2,2nr | head -8
// echo "== 4. what YARN remembers"
// yarn application -list -appStates FINISHED 2>/dev/null | grep "word count" | awk -F'\t' '{print $1 " | " $2 " | " $6 " | " $7}'
// echo "== 5. run again into the same output folder"
// hadoop jar wc.jar WordCount /user/hadoop/data/feedback.txt /out/wc 2>&1 | grep -E "Exception" | head -1 | cut -c1-160
// echo "== 6. wrong input path"
// hadoop jar wc.jar WordCount /user/hadoop/data/feedbak.txt /out/wc2 2>&1 | grep -E "Exception" | head -1 | cut -c1-160

// In the mr-lab folder:
//   python3 make_feedback.py
//   docker cp feedback.txt lab8-namenode-1:/tmp/mr/
//   docker cp WordCount.java lab8-namenode-1:/tmp/mr/
//   docker cp wordcount_lab.sh lab8-namenode-1:/tmp/mr/
//   docker exec lab8-namenode-1 hdfs dfs -put feedback.txt /user/hadoop/data/   (run from /tmp/mr)
//   docker exec lab8-namenode-1 bash /tmp/mr/wordcount_lab.sh

# kill_task.sh (run on your PC, in the lab folder): kill a running map task and watch Hadoop retry it
docker exec lab8-namenode-1 bash -c 'cd /tmp/mr; for i in $(seq 20); do cat feedback.txt; done > feedback_x20.txt
  hdfs dfs -D dfs.blocksize=33554432 -put -f feedback_x20.txt /user/hadoop/data/ 2>/dev/null
  hdfs dfs -rm -r -f -skipTrash /out/wc_big > /dev/null 2>&1'
docker exec -d lab8-namenode-1 bash -c 'hadoop jar /tmp/mr/wc.jar WordCount /user/hadoop/data/feedback_x20.txt /out/wc_big > /tmp/wc_big.log 2>&1'
until docker exec lab8-worker2-1 pgrep -f "java.*YarnChild.*_m_" > /dev/null; do sleep 1; done
sleep 6                                              # let the map task get going
echo "map tasks running on worker2:"
docker exec lab8-worker2-1 pgrep -af "java.*YarnChild.*_m_" | grep -o "attempt_[0-9]*_[0-9]*_m_[0-9]*_[0-9]*" | sort -u
victim=$(docker exec lab8-worker2-1 pgrep -f "java.*YarnChild.*_m_" | head -1)
docker exec lab8-worker2-1 kill -9 $victim && echo "killed JVM $victim on worker2 at $(date +%T)"
until docker exec lab8-namenode-1 grep -qE "completed successfully|failed with state" /tmp/wc_big.log; do sleep 2; done
docker exec lab8-namenode-1 bash -c 'grep -E "number of splits|map [0-9]+%|Task Id|Error:|exit code|completed|Failed map tasks|Launched map tasks|Data-local|Rack-local" /tmp/wc_big.log | sed -E "s/^[0-9-]+ [0-9:]+ INFO  //" | cut -c1-150
  hdfs dfs -cat /out/wc_big/part-r-00000 | grep -wE "chai|was"
  echo "-- every attempt of this job"
  bash /tmp/mr/attempts.sh $(grep -o "job_[0-9_]*" /tmp/wc_big.log | head -1)'

# Needs attempts.sh (Lesson 8.1) in /tmp/mr in the namenode container.
# Run on your PC, in the mr-lab folder:  bash kill_task.sh

Tokenizing one feedback line

Follow how the tokenizer turns a line with capitals and punctuation into clean words. This is the only logic in the mapper.

Worked examples: real problems, solved and traced in memory

Three end-to-end problems: sizing the shuffle, reading counters after a failure, and planning reruns.

1,033,238 records · 9,226,914 bytes · 11,293,396 materialized

10.9 bytes per record shuffled, 8.9 raw
framing overhead: 2,066,482 bytes (2.0 per record)

3 splits · 5 launched · 1 failed · 1 killed

extra attempts: 2 = 1 failed + 1 killed
map outputs used by the reducer: 3

dates · existing folders

/out/wc/dt=2026-10-19 -> skip (already done)
/out/wc/dt=2026-10-20 -> delete half-written folder, then run
/out/wc/dt=2026-10-21 -> run

Check yourself

Quick revision

Interview and viva questions

Practice problems

Word count exercises using the real counters and output from our lab.

1,033,238 / 200,000

5.17 words per line

print(round(1033238 / 200000, 2), "words per line")

chai 33431 · was 200000

True True

print(33431 * 20 == 668620, 200000 * 20 == 4000000)

137 · 143

137 -> signal 9 SIGKILL
143 -> signal 15 SIGTERM

for code in (137, 143):
    sig = code - 128
    print(code, "-> signal", sig, {9: "SIGKILL", 15: "SIGTERM"}[sig])

8 good words · 7 bad words

140137 59863 70.1% positive

good = {"tasty": 17338, "hot": 17294, "fresh": 17426, "crispy": 17585, "perfect": 17706, "good": 17616, "nice": 17485, "amazing": 17687}
bad = {"cold": 8452, "late": 8792, "oily": 8444, "salty": 8541, "costly": 8410, "stale": 8617, "slow": 8607}
g, b = sum(good.values()), sum(bad.values())
print(g, b, f"{g / (g + b):.1%} positive")

one line

['loved', 'the', 'vada pav', 'was', 'tasty']

PHRASES = {"vada pav"}
def tokens(line):
    w = line.lower().split()
    out, i = [], 0
    while i < len(w):
        if i + 1 < len(w) and w[i] + " " + w[i + 1] in PHRASES:
            out.append(w[i] + " " + w[i + 1]); i += 2
        else:
            out.append(w[i]); i += 1
    return out
print(tokens("loved the vada pav was tasty"))

4 reducers · 31 words

4 part-r files: part-r-00000 .. part-r-00003
each word goes to exactly one reducer: (hash(word) & MAX) % 4

print(4, "part-r files: part-r-00000 .. part-r-00003")
print("each word goes to exactly one reducer: (hash(word) & MAX) % 4")

2,000 tasks · 60 s · 1% retries

20 extra task-minutes per night (1% of the job)

extra = 2000 * 0.01 * 60
print(f"{extra / 60:.0f} extra task-minutes per night ({extra / (2000 * 60):.0%} of the job)")

3 splits

shuffled without combiner: 20 | with combiner: 15

splits = [["the chai was hot", "the chai was cold"], ["the dosa was hot"], ["the chai was late", "the idli was hot"]]
plain = combined = 0
for split in splits:
    words = [w for line in split for w in line.split()]
    plain += len(words)
    combined += len(set(words))
print("shuffled without combiner:", plain, "| with combiner:", combined)

Real-world scenario questions (asked in company interviews)

Interviewers love asking you to walk through a job end to end, and what happens when it breaks.