The Mapper

1 The story

Three hundred bad lines in three lakh

The canteen's new billing tablets have bugs. Kabir finds the export full of surprises: an order with no amount, a samosa that cost minus 18 rupees, “Vada Pav ” typed with capitals and a space, a header line repeated in the middle of the file, a price written as “abc”. Riya's revenue job from Lesson 8.1 would crash on the first “abc”. She needs a cleaning step that keeps the good orders, repairs what can be repaired, and counts everything it throws away, so the canteen manager can see how bad the tablets are.

Arjun Sir

Cleaning is the mapper's favourite job. Each line can be judged on its own, so it fits map perfectly, and nothing needs to be grouped, so we don't even need a reducer.

Riya

But if I just skip bad lines, nobody will know how many there were.

Kabir

That's what counters are for. Every map task adds to them, Hadoop totals them for the whole job, and they show up right under the job output. The manager gets exact numbers without anyone writing a report.

Arjun Sir

Today we look inside the Mapper class itself: its four type parameters, the setup–map–cleanup lifecycle, the Context object, and a few habits that separate a fast mapper from a slow or wrong one.

2 Definition

Formal definition (write this in exams)

In Hadoop's org.apache.hadoop.mapreduce API, a Mapper is a class extending Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>. For every input split the framework creates one map task, instantiates the Mapper, and calls its run() method, which calls setup(context) once, then map(key, value, context) once for every record produced by the split's RecordReader, and finally cleanup(context) once. The map method may emit zero, one or many intermediate key-value pairs with context.write(); their types must match KEYOUT and VALUEOUT and be Hadoop Writable types. The Context also gives access to the job configuration, the input split, counters and status reporting. When the number of reduce tasks is zero (a map-only job), map output is written directly to the output format without being sorted or shuffled.

In simple words

A Mapper is a small class Hadoop runs once per piece of the input. It gets one line at a time, decides what (if anything) to send on, and can count things along the way. Map tasks never share variables, so each one only knows about its own piece.

Words you'll meet in this lesson

KEYIN, VALUEINInput types; for text files LongWritable (byte offset) and Text (the line).

KEYOUT, VALUEOUTTypes of the pairs you emit, e.g. Text, IntWritable.

WritableHadoop's compact, reusable serializable types (Text, IntWritable…).

ContextYour link to the framework: write, getCounter, getConfiguration, getInputSplit.

setup / cleanupCalled once per map task, before the first and after the last record.

CounterA named number summed across all tasks of a job.

Map-only jobsetNumReduceTasks(0): output files part-m-NNNNN, no shuffle.

NullWritableAn empty placeholder key or value that takes zero bytes.

From our lab (Hadoop 3.5.0, dirty_orders.csv stored with 2 MB blocks)

Input	300,302 lines: 3 lakh real orders plus 301 bad or odd lines added by `make_dirty.py`; 3 splits → 3 map tasks
Counters	GOOD=300059 · HEADER=42 · WRONG_FIELDS=94 · BAD_AMOUNT=107 (sum 300,302 = Map input records)
Lifecycle	Each task's log has exactly one `setup()` and one `cleanup()` line; `map()` was called 137,786 + 131,050 + 31,466 = 300,302 times
Map-only output	`part-m-00000` 2,094,939 B · `part-m-00001` 2,095,388 B · `part-m-00002` 502,829 B; Spilled Records=0 (nothing sorted)
Repair	“vada pav,20” appears 50,045 times: 49,986 clean + 59 fixed from “Vada Pav ”

The Mapper at three levels

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

Beginner A ticket checker at the gate

Picture a checker at a college fest gate. People come one by one (records). For each one she decides: let in, fix the name on the pass, or turn away.
She has a clicker for each reason she turns someone away (counters). At the end, the clickers from all gates are added up.
Before the first visitor she sets up her table (setup); after the last she packs up (cleanup). Each gate (map task) works on its own queue and never talks to the others.

Intermediate The class and its lifecycle

class CleanMapper extends Mapper<LongWritable, Text, NullWritable, Text>: in = (byte offset, line), out = (nothing, cleaned line).
Hadoop's built-in run() is: setup(ctx); while (ctx.nextKeyValue()) map(ctx.getCurrentKey(), ctx.getCurrentValue(), ctx); cleanup(ctx);. You override the three methods you need.
map() can return without writing (filter), write once (transform) or write many times (word count writes one pair per word).
A field in the Mapper (like our calls counter) lives for one task only. Three tasks gave three separate counts in the logs; only counters are totalled for the whole job.

Advanced Writing mappers that are fast and correct

Reuse output objects: create one Text as a field and set() it each time instead of new Text() per record. context.write() serializes immediately, so reuse is safe and saves garbage collection over crores of records.
The input value is reused by Hadoop: the same Text object holds every line. If you store it in a list, copy it (new Text(value)) or you'll keep many references to one object (Lesson 8.5 shows this bug in a reducer).
Skip or fail? Skip and count bad records when the data is expected to be messy. Throw an exception when bad data means something is badly wrong: a task that throws is retried (4 attempts by default) and then fails the job.
Expensive setup goes in setup(): reading a lookup file, compiling a regex, opening a connection. Doing it in map() repeats it 3 lakh times.
Map output is what gets sorted and shuffled: emit only the fields you need.

Exam tip: For “Explain the Mapper class lifecycle”, write run() = setup() once → map() once per record → cleanup() once, state that there is one Mapper instance per map task (per input split), and give one use of each: setup to load configuration, map to transform or filter, cleanup to emit totals or close resources.

3 Types and ways to use it

Writable types and Context methods

Writable	Java type	Typical use
`Text`	String (UTF-8)	Lines, words, item names
`IntWritable` / `LongWritable`	int / long	Counts, amounts; LongWritable is the byte-offset key of text input
`DoubleWritable` / `FloatWritable`	double / float	Averages, measurements
`BooleanWritable`	boolean	Flags
`NullWritable`	nothing (singleton)	When you need only a key or only a value
`ArrayWritable`, `MapWritable`, custom	collections, your own class	Several fields in one value (custom types: Lesson 9.3)

What you do through the Context

context.write(k, v)

Emit an intermediate pair (or, in a map-only job, an output record). Call it 0, 1 or many times per record.

output

context.getCounter(…)

Increment a counter, from an enum (Quality.BAD_AMOUNT) or a group and name string.

data quality

context.getConfiguration()

Read job settings passed with -D name=value, e.g. a threshold or a date.

parameters

context.getInputSplit()

Which file and byte range this task reads; our log printed dirty_orders.csv:0+2097152.

debugging

Kinds of mappers

Mapper does	Pairs per record	Example
Filter	0 or 1	Keep valid orders (this lesson)
Transform	exactly 1	(item, amount) from a CSV line (Lesson 8.1)
Explode	many	(word, 1) for every word (Lesson 8.6)
Aggregate in the task	a few, in cleanup()	Keep totals in a HashMap, emit them at the end (“in-mapper combining”)

A job with zero reducers is often the fastest job you can write: no sort, no spill, no shuffle. Our cleaning job reported Spilled Records=0. Filtering, format conversion and data-quality checks are usually map-only.

4 Step by step

A cleaning mapper on dirty data

Run on the Module 8 lab with mapper_lab.sh (in “Run it on your PC”).

Declare the types and the counters

enum Quality { GOOD, HEADER, WRONG_FIELDS, BAD_AMOUNT }

public static class CleanMapper
    extends Mapper<LongWritable, Text, NullWritable, Text> {
  private final Text out = new Text();     // reused for every record
  private long calls = 0;                  // lives for ONE task only

Input: (byte offset, line). Output: no key, just the cleaned line, so the key is NullWritable. An enum gives counter names that can't be mistyped.

Judge each record in map()

String[] f = line.toString().split(",");
if (f[0].equals("order_id")) { ctx.getCounter(Quality.HEADER).increment(1); return; }
if (f.length != 3) { ctx.getCounter(Quality.WRONG_FIELDS).increment(1); return; }
try { amount = Integer.parseInt(f[2].trim()); }
catch (NumberFormatException e) { ctx.getCounter(Quality.BAD_AMOUNT).increment(1); return; }
if (amount <= 0) { ctx.getCounter(Quality.BAD_AMOUNT).increment(1); return; }
ctx.getCounter(Quality.GOOD).increment(1);
out.set(f[0] + "," + f[1].trim().toLowerCase() + "," + amount);
ctx.write(NullWritable.get(), out);

Every bad case returns early without writing, after counting why. “abc” is caught instead of crashing the task. The item name is trimmed and lower-cased, which repairs “Vada Pav ”.

Make it map-only and run it
```
hdfs dfs -D dfs.blocksize=2097152 -put -f dirty_orders.csv /user/hadoop/data/
hadoop jar mr.jar CleanOrders /user/hadoop/data/dirty_orders.csv /out/clean
```
number of splits:3 Job job_1792601400460_0016 completed successfully Launched map tasks=3 Map input records=300302 Map output records=300059 Spilled Records=0
The driver calls job.setNumReduceTasks(0). We stored the file with 2 MB blocks to get 3 map tasks. No reducer means nothing is sorted or spilled.
Read the custom counters
```
	CleanOrders$Quality
		BAD_AMOUNT=107
		GOOD=300059
		HEADER=42
		WRONG_FIELDS=94
```
Hadoop added our enum as its own counter group. 107 = 46 “abc” + 61 negative amounts; 94 = 47 lines with two fields + 47 with four; 42 = the real header + 41 repeated ones. Together: 300,302, every input line accounted for.
Look at the output files
```
hdfs dfs -ls /out/clean
hdfs dfs -cat /out/clean/part-m-* | grep -c "vada pav,20"
```
0 /out/clean/_SUCCESS 2094939 /out/clean/part-m-00000 2095388 /out/clean/part-m-00001 502829 /out/clean/part-m-00002 50045
One part-m file per map task (m for map), in input order, not sorted. 50,045 vada pav orders = 49,986 originally clean + 59 repaired.
See the lifecycle in the task logs
```
yarn logs -applicationId application_1792601400460_0016 -log_files stderr
```
Container: container_…_01_000003 on worker1_35119 setup() once, before the first record. split = …dirty_orders.csv:2097152+2097152 cleanup() once, after the last record. map() was called 131050 times Container: container_…_01_000001 on worker2_46619 Container: container_…_01_000004 on worker3_45331 setup() once, before the first record. split = …dirty_orders.csv:4194304+503302 cleanup() once, after the last record. map() was called 31466 times Container: container_…_01_000002 on worker3_45331 setup() once, before the first record. split = …dirty_orders.csv:0+2097152 cleanup() once, after the last record. map() was called 137786 times
Three map tasks, three Mapper objects, each with its own calls field. Container _000001 is the MRAppMaster, so it printed nothing. 137,786 + 131,050 + 31,466 = 300,302.

Don't use a static or instance field to count across the whole job: each map task runs in its own JVM, often on another machine, so fields are never shared. Use counters for job-wide numbers, and keep counters to a few dozen names; they are held in memory by the MRAppMaster (the default limit is 120 counters).

5 See it happen

Three map tasks, one lifecycle each

Six steps of the cleaning job: setup, records flowing through map(), counters, cleanup and the final totals.

setup → map → cleanup 6 steps

What you see

Behind the screen

6 Real-world use

Mappers at work

1. Landing-zone cleaning

The first job in many pipelines is map-only: validate each record from source systems, standardise formats, and drop or quarantine bad rows while counters feed a data-quality dashboard.

Raw files→Map-only validation→Clean zone + counts

2. Counters as pipeline gates

The driver reads counters after waitForCompletion() and stops the pipeline when too many records are bad, instead of loading half-broken data into reports.

Job done→job.getCounters()→Continue or alert

3. Format conversion and masking

Map-only jobs convert text into compact formats and mask personal data (for example the middle digits of a phone number) before analysts can see it.

Text logs→Map: parse + mask→Avro / Parquet

4. Enrichment with a lookup table

A small reference file (store codes → city) is loaded into a HashMap in setup(), and every record is enriched in map() with no reducer. Lesson 9.6 builds on this with the distributed cache.

Records→setup: load lookup · map: add city→Enriched records

Simplified; organisations differ in tools and policies.

7 Run the code

Simulate a Mapper with setup, map, cleanup and counters

This Python class mirrors CleanOrders.CleanMapper. The run() method is what Hadoop's own Mapper.run() does: setup once, map per record, cleanup once. Counters are a dictionary here.

class CleanMapper:
    def setup(self, ctx):
        self.calls = 0
        print("setup() for split", ctx["split"])

    def map(self, key, line, ctx):
        self.calls += 1
        f = line.split(",")
        if f[0] == "order_id":
            ctx["counters"]["HEADER"] += 1; return
        if len(f) != 3:
            ctx["counters"]["WRONG_FIELDS"] += 1; return
        try:
            amount = int(f[2].strip())
        except ValueError:
            ctx["counters"]["BAD_AMOUNT"] += 1; return
        if amount <= 0:
            ctx["counters"]["BAD_AMOUNT"] += 1; return
        ctx["counters"]["GOOD"] += 1
        ctx["out"].append(f"{f[0]},{f[1].strip().lower()},{amount}")

    def cleanup(self, ctx):
        print("cleanup(): map() was called", self.calls, "times")

    def run(self, records, ctx):                 # what Hadoop's Mapper.run() does
        self.setup(ctx)
        for key, value in records:
            self.map(key, value, ctx)
        self.cleanup(ctx)

lines = ["order_id,item,amount", "1001,samosa,18", "901714,chai,abc", "1002,vada pav,20",
         "904721,Vada Pav ,20", "905000,samosa,-18", "906000,dosa", "907000,poha,25,extra"]
ctx = {"split": "dirty_orders.csv:0+2097152", "out": [],
       "counters": {"GOOD": 0, "HEADER": 0, "WRONG_FIELDS": 0, "BAD_AMOUNT": 0}}
offset, records = 0, []
for l in lines:                                   # TextInputFormat: key = byte offset
    records.append((offset, l))
    offset += len(l) + 1
CleanMapper().run(records, ctx)
print(ctx["counters"])
print(ctx["out"])

setup() for split dirty_orders.csv:0+2097152 cleanup(): map() was called 8 times {'GOOD': 3, 'HEADER': 1, 'WRONG_FIELDS': 2, 'BAD_AMOUNT': 2} ['1001,samosa,18', '1002,vada pav,20', '904721,vada pav,20']

def run(self, records, ctx): # what Hadoop's Mapper.run() doesHadoop calls run(); the default runs setup, then map for every record, then cleanup. You rarely override it.
ctx["counters"]["GOOD"] += 1Like ctx.getCounter(Quality.GOOD).increment(1): Hadoop sums these over all tasks of the job.
ctx["out"].append(f"{f[0]},{f[1].strip().lower()},{amount}")Like context.write(): trim and lower-case repairs “Vada Pav ” into “vada pav”.

Run it on your PC: clean dirty orders with a map-only job

Use the Module 8 lab from Lesson 8.1. First create the dirty file next to canteen_orders.csv with make_dirty.py, then copy both files below into /tmp/mr in the namenode container, compile and run mapper_lab.sh.

Make the dirty data

# make_dirty.py: copy canteen_orders.csv and break some lines on purpose (same result every time)
import random
random.seed(3)
lines = open("canteen_orders.csv").read().splitlines()
bad = ["{id},dosa", "{id},chai,abc", "{id},samosa,-18", "{id},poha,25,extra", "{id},Vada Pav ,20", "order_id,item,amount"]
out = [lines[0]]
counts = {}
for line in lines[1:]:
    out.append(line)
    if random.random() < 0.001:                 # about 1 line in 1,000
        b = random.choice(bad)
        counts[b] = counts.get(b, 0) + 1
        out.append(b.format(id=900000 + len(out)))
open("dirty_orders.csv", "w").write("\n".join(out) + "\n")
print(len(out), "lines;", counts)

# Run, next to canteen_orders.csv (from Lesson 4.3):
#   python3 make_dirty.py
#   docker cp dirty_orders.csv lab8-namenode-1:/tmp/mr/

300302 lines; {'{id},chai,abc': 46, '{id},poha,25,extra': 47, '{id},dosa': 47, 'order_id,item,amount': 41, '{id},Vada Pav ,20': 59, '{id},samosa,-18': 61}

Real output. Seeded, so you get exactly the same 301 extra lines. These numbers explain every counter in the next step.

The cleaning mapper and the lab run

// CleanOrders.java: a map-only job. Keeps good orders, counts bad ones with custom counters,
// and logs the Mapper lifecycle (setup -> map ... -> cleanup) to the task log.
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class CleanOrders {
  enum Quality { GOOD, HEADER, WRONG_FIELDS, BAD_AMOUNT }

  public static class CleanMapper extends Mapper<LongWritable, Text, NullWritable, Text> {
    private final Text out = new Text();
    private long calls = 0;

    @Override
    protected void setup(Context ctx) {
      System.err.println("setup()   once, before the first record. split = " + ctx.getInputSplit());
    }

    @Override
    protected void map(LongWritable offset, Text line, Context ctx) throws IOException, InterruptedException {
      calls++;
      String[] f = line.toString().split(",");
      if (f[0].equals("order_id")) { ctx.getCounter(Quality.HEADER).increment(1); return; }
      if (f.length != 3) { ctx.getCounter(Quality.WRONG_FIELDS).increment(1); return; }
      int amount;
      try { amount = Integer.parseInt(f[2].trim()); }
      catch (NumberFormatException e) { ctx.getCounter(Quality.BAD_AMOUNT).increment(1); return; }
      if (amount <= 0) { ctx.getCounter(Quality.BAD_AMOUNT).increment(1); return; }
      ctx.getCounter(Quality.GOOD).increment(1);
      out.set(f[0] + "," + f[1].trim().toLowerCase() + "," + amount);
      ctx.write(NullWritable.get(), out);
    }

    @Override
    protected void cleanup(Context ctx) {
      System.err.println("cleanup() once, after the last record. map() was called " + calls + " times");
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
    Job job = Job.getInstance(conf, "clean orders");
    job.setJarByClass(CleanOrders.class);
    job.setMapperClass(CleanMapper.class);
    job.setNumReduceTasks(0);
    job.setOutputKeyClass(NullWritable.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);
  }
}


// ---- mapper_lab.sh ----
// # mapper_lab.sh: run the map-only CleanOrders job on dirty data and look at counters, output files and task logs
// cd /tmp/mr
// hdfs dfs -D dfs.blocksize=2097152 -put -f dirty_orders.csv /user/hadoop/data/
// hdfs dfs -rm -r -f -skipTrash /out/clean > /dev/null 2>&1
// hadoop jar mr.jar CleanOrders /user/hadoop/data/dirty_orders.csv /out/clean 2>&1 \
//   | grep -E "number of splits|Submitted application|map 100%|completed|Launched|Map input records|Map output records|Spilled|CleanOrders|GOOD|HEADER|WRONG_FIELDS|BAD_AMOUNT"
// echo "--- output files (map-only: part-m-*)"
// hdfs dfs -ls /out/clean | awk '{print $5, $8}'
// hdfs dfs -cat /out/clean/part-m-00000 2>/dev/null | head -3
// hdfs dfs -cat /out/clean/part-m-* | grep -c "vada pav,20"     # 49,986 clean + 59 fixed from "Vada Pav "
// echo "--- task logs (stderr of each map task)"
// app=$(yarn application -list -appStates FINISHED 2>/dev/null | grep "clean orders" | awk '{print $1}' | tail -1)
// sleep 5
// yarn logs -applicationId $app -log_files stderr 2>/dev/null | grep -E "^Container: |setup\(\)|cleanup\(\)"

// In the mr-lab folder:
//   docker cp CleanOrders.java lab8-namenode-1:/tmp/mr/
//   docker cp mapper_lab.sh lab8-namenode-1:/tmp/mr/
//   docker exec lab8-namenode-1 bash -c 'cd /tmp/mr && mkdir -p classes && javac -cp $(hadoop classpath) -d classes *.java && jar cf mr.jar -C classes .'
//   docker exec lab8-namenode-1 bash /tmp/mr/mapper_lab.sh

2026-10-21 17:01:45 INFO JobSubmitter:203 - number of splits:3 2026-10-21 17:01:46 INFO YarnClientImpl:356 - Submitted application application_1792601400460_0016 2026-10-21 17:02:02 INFO Job:1754 - map 100% reduce 0% 2026-10-21 17:02:02 INFO Job:1765 - Job job_1792601400460_0016 completed successfully Launched map tasks=3 Map input records=300302 Map output records=300059 Spilled Records=0 CleanOrders$Quality BAD_AMOUNT=107 GOOD=300059 HEADER=42 WRONG_FIELDS=94 --- output files (map-only: part-m-*) 0 /out/clean/_SUCCESS 2094939 /out/clean/part-m-00000 2095388 /out/clean/part-m-00001 502829 /out/clean/part-m-00002 1001,samosa,18 1002,vada pav,20 1003,poha,25 50045 --- task logs (stderr of each map task) Container: container_1792601400460_0016_01_000003 on worker1_35119 setup() once, before the first record. split = hdfs://namenode/user/hadoop/data/dirty_orders.csv:2097152+2097152 cleanup() once, after the last record. map() was called 131050 times Container: container_1792601400460_0016_01_000001 on worker2_46619 Container: container_1792601400460_0016_01_000004 on worker3_45331 setup() once, before the first record. split = hdfs://namenode/user/hadoop/data/dirty_orders.csv:4194304+503302 cleanup() once, after the last record. map() was called 31466 times Container: container_1792601400460_0016_01_000002 on worker3_45331 setup() once, before the first record. split = hdfs://namenode/user/hadoop/data/dirty_orders.csv:0+2097152 cleanup() once, after the last record. map() was called 137786 times

Real output. Container _000001 is the MRAppMaster, which is why it has no setup() or cleanup() lines. Which worker runs which task will differ on your PC.

8 Line by line

Mapper.run(): one setup, many maps, one cleanup

Hadoop's Mapper.run() in a few lines of Python. Watch calls grow and bad count the skipped lines.

Worked examples: real problems, solved and traced in memory

Three Mapper problems: picking the type parameters, deciding skip vs fail, and making a slow mapper fast.

Example 1BeginnerTypes · generics

Choose the four type parameters

Problem statement

A mapper reads text lines student_id,canteen,amount and emits (canteen, amount). Write its Mapper<…> declaration and show what one line becomes.

Input

TextInputFormat · line “S17,north,45”

Expected output

Mapper<LongWritable, Text, Text, IntWritable>
(0, 'S17,north,45') -> ('north', 45)

Approach

Text input gives (byte offset, line): LongWritable, Text.
Output key is a canteen name: Text. Output value is an amount: IntWritable.
Simulate one record.

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 2IntermediateDesign · bad data

Skip and count, or fail the job?

Problem statement

A job loads 10 crore bank transactions a night. Policy: up to 0.01% bad rows may be skipped; above that the load must stop. The mapper counts BAD rows. How does the driver enforce this, and is our canteen file (243 bad of 300,302) acceptable under the same policy?

Input

limit 0.01% · canteen: 243 / 300,302

Expected output

bad = 0.081% -> STOP the pipeline

Approach

Mappers skip and count; they cannot see the job total.
After waitForCompletion(), the driver reads the counters and decides.
Compute the canteen ratio.

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 3AdvancedPerformance · reuse

Why setup() and object reuse matter

Problem statement

A mapper compiles a regex and creates a new Text for every record. Compiling costs 20 µs, creating an object 0.2 µs plus garbage-collection work. For 2.7 crore records, how much time goes to work that could be done once?

Input

27,000,000 records · 20 µs + 0.2 µs per record

Expected output

in map(): 545 s of CPU; in setup(): 0.00008 s

Approach

Per-record waste × records.
Move the compile to setup() and reuse one Text field.
Compare with doing it once per task (4 tasks).

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.3 in one screen

Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>; text input = (LongWritable offset, Text line).
run() = setup() once → map() per record → cleanup() once; one Mapper object per map task.
context.write() 0, 1 or many times; getCounter, getConfiguration, getInputSplit.
Writables: Text, IntWritable, LongWritable, DoubleWritable, NullWritable; reuse your output objects, copy input values before storing.
Fields are per task; job-wide numbers need counters. Lab: GOOD 300,059 + HEADER 42 + WRONG_FIELDS 94 + BAD_AMOUNT 107 = 300,302.
Map-only: setNumReduceTasks(0) → part-m-NNNNN, no sort, Spilled Records=0.
Expected bad data: skip and count; unexpected: throw (task retried, then job fails). Heavy work in setup().

Interview and viva questions

What is the Mapper in Hadoop MapReduce?

The Mapper is the user class that processes input records. Hadoop runs one map task per input split; inside it, the Mapper's setup() is called once, map() once for every key-value record from the RecordReader, and cleanup() once. map() emits intermediate key-value pairs through the Context, which are then partitioned, sorted and shuffled to reducers (or written directly as output in a map-only job).

What are setup() and cleanup() used for in a Mapper?

setup() runs once per map task before any record: use it to read configuration values, load small lookup files, compile patterns or open resources. cleanup() runs once after the last record: use it to emit results aggregated inside the task (in-mapper combining) and to close resources.

What is a map-only job?

A job with the number of reduce tasks set to 0 (job.setNumReduceTasks(0)). The output of each map task is written directly by the OutputFormat as part-m-NNNNN files, skipping the sort, spill, shuffle and reduce phases. It suits filtering, cleaning, format conversion and enrichment.

How do you create custom counters in MapReduce?

Define a Java enum (enum Quality { GOOD, BAD }) and call context.getCounter(Quality.GOOD).increment(1), or use context.getCounter("group", "name"). The framework sums counters from all task attempts that succeed and shows them in the job output, the web UI and job.getCounters() in the driver.

Why does Hadoop use Writable types instead of Java types?

Writable types (Text, IntWritable, LongWritable…) serialize compactly to a byte stream with write() and readFields(), are fast to compare in their serialized form during the sort, and can be reused and refilled with set() to avoid creating objects for every record. Keys must be WritableComparable so they can be sorted.

Practice problems

Mapper exercises built on the real counters and logs from the lab.

P1Easy

Every line accounted for

Check that the four counters add up to Map input records.

Input

GOOD 300059 · HEADER 42 · WRONG_FIELDS 94 · BAD_AMOUNT 107

Expected output

True

Show solution and explanation

print(300059 + 42 + 94 + 107 == 300302)

True

How it works: True: each line hits exactly one counter.

P2Easy

Explain BAD_AMOUNT

make_dirty.py added 46 “abc” amounts and 61 negative amounts. Do they explain BAD_AMOUNT=107?

Input

46 + 61

Expected output

Show solution and explanation

print(46 + 61)

107

How it works: 107: both kinds fail the amount check, one in parseInt, one in the “≤ 0” test.

P3Easy

map() calls per task

Add the three map() call counts from the task logs.

Input

137,786 · 131,050 · 31,466

Expected output

Show solution and explanation

print(137786 + 131050 + 31466)

300302

How it works: 300,302, equal to Map input records: map() is called once per record, in exactly one task.

P4Medium

Repair counted

Without the trim/lower-case repair, how many “vada pav,20” lines would be in the output, and which counter would the 59 “Vada Pav ” lines land in?

Input

50,045 with repair · 59 repaired

Expected output

49986 lines; the 59 would still be GOOD, just with a different item name

Show solution and explanation

print(50045 - 59, "lines; the 59 would still be GOOD, just with a different item name")

49986 lines; the 59 would still be GOOD, just with a different item name

How it works: 49,986. They would pass all checks and be written as “Vada Pav ”, creating a seventh “item” in later reports. Counters only catch what you test for.

P5Medium

Output size per task

The three part-m files are 2,094,939, 2,095,388 and 502,829 bytes. Why is the last one about a quarter of the others?

Input

split lengths 2 MB, 2 MB, 0.48 MB

Expected output

0.23999309539794922

Show solution and explanation

print(503302 / 2097152)

0.23999309539794922

How it works: Its split is only 503,302 bytes (0.24 of a 2 MB split), the leftover end of the file.

P6Medium

In-mapper totals

Instead of emitting (item, amount) for every order, keep totals in a dictionary and emit them once in cleanup(). How many pairs leave this task?

Input

one task, 6 orders

Expected output

chai 45
dosa 90
poha 25
6 records -> 3 pairs emitted

Show solution and explanation

orders = [("chai", 15), ("dosa", 45), ("chai", 15), ("poha", 25), ("dosa", 45), ("chai", 15)]
totals = {}
for item, amount in orders:                 # map()
    totals[item] = totals.get(item, 0) + amount
for item, total in totals.items():          # cleanup()
    print(item, total)
print(len(orders), "records ->", len(totals), "pairs emitted")

chai 45 dosa 90 poha 25 6 records -> 3 pairs emitted

How it works: 3 pairs instead of 6. With 6 items and 9 lakh orders per task, 6 pairs instead of 9 lakh, which shrinks the shuffle enormously (combiners do this automatically, Lesson 9.1).

P7Business problem

Is the tablet vendor good enough?

The contract says at most 1 bad order line per 1,000. Our file has 243 bad lines among 300,302 (ignoring the one true header). Does the vendor meet it?

Input

242 bad of 300,301 data lines

Expected output

0.81 bad lines per 1,000 -> OK

Show solution and explanation

rate = (243 - 1) / 300301 * 1000
print(f"{rate:.2f} bad lines per 1,000 ->", "OK" if rate <= 1 else "breach")

0.81 bad lines per 1,000 -> OK

How it works: 0.81 per 1,000: within the contract, barely. Counters give the manager this number every night for free.

P8Advanced

Mapper with a lookup in setup()

Load a price list once in setup() and flag orders whose amount does not match the menu price, using a counter.

Input

menu + 5 orders

Expected output

{'OK': 3, 'WRONG_PRICE': 2}

Show solution and explanation

class PriceCheck:
    def setup(self):
        self.menu = {"chai": 15, "samosa": 18, "dosa": 45}      # e.g. read from a small file
        self.counters = {"OK": 0, "WRONG_PRICE": 0}
    def map(self, line):
        oid, item, amount = line.split(",")
        key = "OK" if self.menu.get(item) == int(amount) else "WRONG_PRICE"
        self.counters[key] += 1
m = PriceCheck(); m.setup()
for line in ["1,chai,15", "2,dosa,40", "3,samosa,18", "4,chai,20", "5,dosa,45"]:
    m.map(line)
print(m.counters)

{'OK': 3, 'WRONG_PRICE': 2}

How it works: OK 3, WRONG_PRICE 2. The menu is loaded once per task, not once per line; Lesson 9.6 ships such files to every task with the distributed cache.

Real-world scenario questions (asked in company interviews)

Mapper questions test whether you have written real MapReduce code: lifecycle, types, counters and the reuse trap.

Concepts · lifecycleFresher

“What methods does a Mapper have, and when are they called?”

A standard question after “explain MapReduce”.

The interviewer is checking: setup, map, cleanup, run, and one object per task.

Show a strong answer

run() drives the task: setup() once, map() for every record of the split, cleanup() once.
One Mapper instance per map task; tasks don't share fields.
setup for configuration and lookups, cleanup for emitting in-task totals and closing resources.

Data quality · designFresher

The job fails with NumberFormatException after 40 minutes

One line in 3 crore has “N/A” as the amount.

The interviewer is checking: defensive mappers and counters.

Show a strong answer

Catch the parse error, increment a BAD_AMOUNT counter and skip (or write the line to a side output for review).
In the driver, read the counters and fail the pipeline only if the bad ratio exceeds an agreed threshold.
Throwing in map() makes the task retry 4 times on the same line and then kills the whole job.

Debugging · object reuseExperienced

A mapper that buffers lines outputs the same line again and again

The code adds each value to an ArrayList and emits the list in cleanup().

The interviewer is checking: knowing that Hadoop reuses input objects.

Show a strong answer

The RecordReader refills the same Text object for every record, so the list holds many references to one object, which ends up holding the last line.
Fix: store copies (new Text(value) or value.toString()).
Also check the buffer size: holding a whole split in memory can exceed the container's heap.

Performance · designExperienced

“Our cleaning job has 200 reducers that just write what they receive. Can it be faster?”

An inherited ETL job validates records and passes them through an identity reducer.

The interviewer is checking: map-only jobs.

Show a strong answer

If nothing needs grouping or sorting, set the reducers to 0: no spill, sort or shuffle, often several times faster.
Output becomes one part-m file per map task; if file count matters, tune split size or compact later.
Keep data-quality numbers in counters and check them in the driver.

The Mapper

Three hundred bad lines in three lakh

The Mapper

Words you'll meet in this lesson

From our lab (Hadoop 3.5.0, dirty_orders.csv stored with 2 MB blocks)

The Mapper at three levels

Beginner A ticket checker at the gate

Intermediate The class and its lifecycle

Advanced Writing mappers that are fast and correct

Writable types and Context methods

What you do through the Context

context.write(k, v)

context.getCounter(…)

context.getConfiguration()

context.getInputSplit()

Kinds of mappers

A cleaning mapper on dirty data

Declare the types and the counters

Judge each record in map()

Make it map-only and run it

Read the custom counters

Look at the output files

See the lifecycle in the task logs

Three map tasks, one lifecycle each

Mappers at work

1. Landing-zone cleaning

2. Counters as pipeline gates

3. Format conversion and masking

4. Enrichment with a lookup table

Simulate a Mapper with setup, map, cleanup and counters

Run it on your PC: clean dirty orders with a map-only job

Make the dirty data

The cleaning mapper and the lab run

Mapper.run(): one setup, many maps, one cleanup

Worked examples: real problems, solved and traced in memory

Choose the four type parameters

Skip and count, or fail the job?

Why setup() and object reuse matter

Check yourself

Quick revision

Lesson 8.3 in one screen

Interview and viva questions

Practice problems

Every line accounted for

Explain BAD_AMOUNT

map() calls per task

Repair counted

Output size per task

In-mapper totals

Is the tablet vendor good enough?

Mapper with a lookup in setup()

Real-world scenario questions (asked in company interviews)

“What methods does a Mapper have, and when are they called?”

The job fails with NumberFormatException after 40 minutes

A mapper that buffers lines outputs the same line again and again

“Our cleaning job has 200 reducers that just write what they receive. Can it be faster?”