Unit 3: Map Reduce and YARN - Subjective Questions

The Mapper processes input records and produces intermediate key-value pairs.

The class declaration is:

public class MyMapper extends Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
    @Override
    protected void map(KEYIN key, VALUEIN value, Context context)
            throws IOException, InterruptedException {
        // Mapping logic
        context.write(outputKey, outputValue);
    }
}

Generic parameters:

• KEYIN: Type of the input key.
• VALUEIN: Type of the input value.
• KEYOUT: Type of the intermediate output key.
• VALUEOUT: Type of the intermediate output value.

For a text file, the default input types are generally LongWritable and Text. The key is the byte offset of a line, while the value is the line itself.

Example:

public class LengthMapper
        extends Mapper<LongWritable, Text, Text, IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String word = value.toString().trim();
        context.write(new Text(word), new IntWritable(word.length()));
    }
}

Hadoop uses Writable classes such as Text, IntWritable, and LongWritable because they support efficient serialization.

A mapper task has three important lifecycle methods:

1. setup(Context context)

   • Called exactly once before processing the input split.
   • Used to initialize resources, read configuration values, or load a small lookup file.

2. map(KEYIN key, VALUEIN value, Context context)

   • Called once for every input record.
   • Converts each input record into zero, one, or several intermediate key-value pairs.

3. cleanup(Context context)

   • Called once after all records assigned to the mapper have been processed.
   • Used to release resources or emit mapper-level summary information.

public class SampleMapper
        extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final IntWritable one = new IntWritable(1);

    @Override
    protected void setup(Context context) {
        // Initialize mapper-level resources
    }

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        context.write(value, one);
    }

    @Override
    protected void cleanup(Context context) {
        // Release resources
    }
}

Important: A mapper object may process many records, so reusable output objects can reduce unnecessary object creation. However, state stored in a mapper is local to that mapper task and is not globally shared.

A Reducer receives a key and all intermediate values associated with that key. Its general structure is:

public class MyReducer
        extends Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
    @Override
    protected void reduce(KEYIN key, Iterable<VALUEIN> values,
                          Context context)
            throws IOException, InterruptedException {
        // Reduction logic
        context.write(outputKey, outputValue);
    }
}

Type parameters:

• KEYIN: Intermediate key type produced by the mapper.
• VALUEIN: Intermediate value type produced by the mapper.
• KEYOUT: Final output key type.
• VALUEOUT: Final output value type.

Example: summing integer values:

public class SumReducer
        extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values,
                          Context context)
            throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable value : values) {
            sum += value.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

The reducer is invoked once for each distinct intermediate key. The shuffle and sort phase ensures that all values belonging to the same key are grouped together before reduce() is called.

The mapper and reducer are connected through the following stages:

• Partitioning: Each mapper output key is assigned to a reducer. The default partition is approximately calculated as , where is the number of reducers.
• Shuffle: Intermediate key-value pairs are transferred from mapper nodes to the appropriate reducer nodes.
• Sort: Hadoop sorts intermediate records by key.
• Grouping: All values associated with an equal key are placed into one logical group.
• Reduction: The reducer receives a key and an Iterable containing its values.

For mapper output:

(cat, 1), (dog, 1), (cat, 1)

The reducer receives:

(cat, [1, 1])
(dog, [1])

The reducer can then produce:

(cat, 2)
(dog, 1)

A custom Partitioner controls reducer selection, while grouping and sorting comparators control key ordering and grouping. Correct mapper and reducer type declarations are essential because the mapper output types must match the reducer input types.

Mapper declaration:

Mapper<LongWritable, Text, Text, IntWritable>

Reducer declaration:

Reducer<Text, IntWritable, Text, IntWritable>

A mapper may emit zero, one, or many records for each input record. A reducer aggregates records only after Hadoop has partitioned, shuffled, sorted, and grouped mapper output. A map-only job is possible by setting the number of reducers to zero, but a reduce-only MapReduce job is not normally created because reducer input originates from mapper output.

The driver creates the Hadoop configuration, defines the job components, specifies input and output paths, and submits the job.

public class JobDriver {
    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("Usage: JobDriver <input> <output>");
            System.exit(2);
        }

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "sample job");

        job.setJarByClass(JobDriver.class);
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

Important operations:

• Job.getInstance() creates a job configuration.
• setJarByClass() helps Hadoop locate the application JAR.
• setMapperClass() and setReducerClass() register processing classes.
• Map-output classes describe intermediate data.
• Output classes describe final data.
• waitForCompletion(true) submits the job and waits for its result.

The output directory must normally not exist before job execution.

• InputFormat: Determines how input data is divided and interpreted. The default TextInputFormat treats each line as one record.
• InputSplit: Represents a logical portion of the input assigned to one mapper. It does not normally contain the data itself; it describes where the data is located.
• RecordReader: Converts data in an input split into key-value records supplied to map(). With TextInputFormat, the key is a byte offset and the value is a line of text.
• OutputFormat: Determines how reducer or mapper output is written. The default TextOutputFormat writes keys and values as text.

They may be configured in the driver as follows:

job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

The number of input splits generally influences the number of mapper tasks. An input split is logical, whereas an HDFS block is a physical storage unit. Hadoop attempts data-local execution by scheduling a mapper near the blocks represented by its split.

YARN stands for Yet Another Resource Negotiator. It is Hadoop's cluster resource-management and application-execution layer.

Its main components are:

• ResourceManager: Global authority that manages cluster resources and accepts applications.
• Scheduler: A part of the ResourceManager that allocates containers according to resource availability and scheduling policies.
• ApplicationsManager: Accepts submissions and starts an ApplicationMaster.
• NodeManager: Runs on each worker node, launches containers, monitors resource use, and reports node status.
• ApplicationMaster: Created for each application. It negotiates containers and coordinates that application's tasks.
• Container: A logical allocation of resources such as memory and virtual CPU cores on a node.

YARN separates resource management from MapReduce processing. Consequently, the same Hadoop cluster can support MapReduce and other distributed processing frameworks. This improves scalability, cluster utilization, and flexibility compared with tightly coupling resource management to a single processing model.

A MapReduce job executes on YARN through these steps:

1. The client creates the job configuration and submits the application to the ResourceManager.
2. Job resources, including the application JAR and configuration, are copied to distributed storage.
3. The ResourceManager allocates a container in which the MapReduce ApplicationMaster is started.
4. The ApplicationMaster determines input splits and calculates the required map and reduce tasks.
5. It negotiates additional containers with the ResourceManager's scheduler.
6. Relevant NodeManagers launch task containers.
7. Mapper tasks read input splits and write partitioned intermediate output.
8. Reducers fetch mapper output during shuffle, sort it, group values, and execute reduce().
9. Tasks report progress and status to the ApplicationMaster.
10. Failed tasks may be scheduled again in new containers.
11. Final output is written to the configured file system path.
12. The ApplicationMaster reports completion, releases resources, and terminates.

This design allows YARN to allocate resources dynamically and isolate failures at the task, container, node, or application level.

In MapReduce v1, the JobTracker handled job submission, resource scheduling, task monitoring, and failure recovery. This created a scalability and availability bottleneck.

YARN separates these duties:

• The ResourceManager performs global resource allocation.
• The ApplicationMaster manages one application.
• NodeManagers manage worker nodes and containers.

Containers are more flexible than fixed map and reduce slots because resources can be assigned according to application needs. YARN therefore improves resource utilization and permits frameworks other than MapReduce to use the Hadoop cluster.

The mapper emits (word, 1) for every word, and the reducer adds all counts associated with each word.

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

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] words = value.toString().toLowerCase()
                .split("[^a-z0-9]+");

        for (String word : words) {
            if (!word.isEmpty()) {
                outputWord.set(word);
                context.write(outputWord, ONE);
            }
        }
    }
}

public class WordReducer
        extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values,
                          Context context)
            throws IOException, InterruptedException {
        int total = 0;
        for (IntWritable value : values) {
            total += value.get();
        }
        context.write(key, new IntWritable(total));
    }
}

For the input Big data big, mapper output is (big,1), (data,1), (big,1). After grouping, the reducer receives (big,[1,1]) and (data,[1]), producing (big,2) and (data,1).

A word-count driver can be configured as follows:

Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "word count");

job.setJarByClass(WordCountDriver.class);
job.setMapperClass(WordMapper.class);
job.setCombinerClass(WordReducer.class);
job.setReducerClass(WordReducer.class);

job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

System.exit(job.waitForCompletion(true) ? 0 : 1);

The combiner performs local aggregation of mapper output. For example, ten local pairs (data,1) can become (data,10) before the shuffle.

Addition is suitable because it is:

• Associative:
• Commutative:

A combiner is an optimization and is not guaranteed to execute. Therefore, program correctness must not depend on it. The output types of the combiner must also be compatible with the mapper output and reducer input types.

Each valid even number is emitted under one constant key. The reducer sums all received values.

public class EvenMapper
        extends Mapper<LongWritable, Text, Text, LongWritable> {
    private static final Text TOTAL = new Text("even-sum");

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] tokens = value.toString().trim().split("\\s+");
        for (String token : tokens) {
            if (token.isEmpty()) {
                continue;
            }
            try {
                long number = Long.parseLong(token);
                if (number % 2 == 0) {
                    context.write(TOTAL, new LongWritable(number));
                }
            } catch (NumberFormatException exception) {
                context.getCounter("INPUT", "INVALID_NUMBER").increment(1);
            }
        }
    }
}

public class EvenSumReducer
        extends Reducer<Text, LongWritable, Text, LongWritable> {
    @Override
    protected void reduce(Text key, Iterable<LongWritable> values,
                          Context context)
            throws IOException, InterruptedException {
        long sum = 0;
        for (LongWritable value : values) {
            sum += value.get();
        }
        context.write(key, new LongWritable(sum));
    }
}

For input 2 7 8 11 12, the answer is . The same reducer may be registered as a combiner because addition is associative and commutative.

The program can be improved in the following ways:

• Filter in the mapper: Emit only numbers satisfying . This prevents odd values from entering the shuffle.
• Use a combiner: Local partial sums reduce network traffic. The sum operation is associative and commutative.
• Mapper-local aggregation: A mapper may maintain a local sum and emit one value in cleanup(), although care is required if no valid even value is found.
• Use LongWritable: It supports a larger range than IntWritable, but very large totals can still overflow.
• Handle invalid tokens: Catch NumberFormatException and increment a Hadoop counter rather than failing the task.
• Use a constant key: A constant key creates one logical total, but it also directs the result to one reducer and may form a bottleneck for very large intermediate data.

A scalable alternative is for mappers or combiners to produce partial sums. The final reducer then adds only those partial results. If values can exceed the long range, arbitrary-precision logic such as BigInteger and a suitable serialization method should be used.

A palindrome reads the same from left to right and right to left. Examples include 121, 1331, and level.

This classification does not require aggregation, so it can be implemented as a map-only job.

public class PalindromeMapper
        extends Mapper<LongWritable, Text, Text, Text> {
    private static boolean isPalindrome(String text) {
        int left = 0;
        int right = text.length() - 1;
        while (left < right) {
            if (text.charAt(left) != text.charAt(right)) {
                return false;
            }
            left++;
            right--;
        }
        return true;
    }

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] tokens = value.toString().trim().split("\\s+");
        for (String token : tokens) {
            if (!token.isEmpty()) {
                String result = isPalindrome(token)
                        ? "PALINDROME" : "NOT_PALINDROME";
                context.write(new Text(token), new Text(result));
            }
        }
    }
}

The driver should include:

job.setMapperClass(PalindromeMapper.class);
job.setNumReduceTasks(0);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

The two-pointer test requires time and extra space for a token of length .

For a non-negative integer , factorial is defined as:

By definition, .

Since every input number can be processed independently, a map-only job is appropriate.

public class FactorialMapper
        extends Mapper<LongWritable, Text, Text, Text> {
    private static BigInteger factorial(int number) {
        BigInteger result = BigInteger.ONE;
        for (int i = 2; i <= number; i++) {
            result = result.multiply(BigInteger.valueOf(i));
        }
        return result;
    }

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] tokens = value.toString().trim().split("\\s+");
        for (String token : tokens) {
            try {
                int number = Integer.parseInt(token);
                if (number >= 0) {
                    context.write(new Text(token),
                            new Text(factorial(number).toString()));
                } else {
                    context.getCounter("INPUT", "NEGATIVE_NUMBER").increment(1);
                }
            } catch (NumberFormatException exception) {
                context.getCounter("INPUT", "INVALID_NUMBER").increment(1);
            }
        }
    }
}

BigInteger avoids the rapid overflow that occurs with int or long. Nevertheless, very large inputs require substantial CPU time and produce extremely large output strings.

When calculating each number's factorial, records are independent:

5 → 120
7 → 5040

No values need to be grouped or combined. Therefore, the mapper can produce final output and the driver can use:

job.setNumReduceTasks(0);

This avoids shuffle and sort overhead.

If the requirement is instead to compute the product of all input numbers, the mapper can emit a constant key and each number as a value:

context.write(new Text("product"), new Text(number.toString()));

A reducer can multiply the grouped values using BigInteger. Multiplication of integers is associative and commutative, so partial products can be computed by combiners. However, a combiner must use output types compatible with the mapper and reducer.

A distinction must be maintained between:

• Factorial of each value: For input , output is and .
• Product of all values: For input , output is .

These are different computational requirements.

An Armstrong number is a non-negative integer equal to the sum of its digits, each raised to the number of digits. For a -digit number with digits :

For example, .

public class ArmstrongMapper
        extends Mapper<LongWritable, Text, LongWritable, Text> {
    private static boolean isArmstrong(long number) {
        if (number < 0) {
            return false;
        }

        String digits = Long.toString(number);
        int power = digits.length();
        BigInteger sum = BigInteger.ZERO;

        for (char character : digits.toCharArray()) {
            int digit = character - '0';
            sum = sum.add(BigInteger.valueOf(digit).pow(power));
        }

        return sum.equals(BigInteger.valueOf(number));
    }

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        for (String token : value.toString().trim().split("\\s+")) {
            try {
                long number = Long.parseLong(token);
                String result = isArmstrong(number)
                        ? "ARMSTRONG" : "NOT_ARMSTRONG";
                context.write(new LongWritable(number), new Text(result));
            } catch (NumberFormatException exception) {
                context.getCounter("INPUT", "INVALID_NUMBER").increment(1);
            }
        }
    }
}

This can be configured as a map-only job because each number is tested independently.

Common design features:

• Each input token can be processed independently.
• The mapper performs parsing, validation, and computation.
• Invalid input can be recorded through counters.
• The driver can call job.setNumReduceTasks(0).
• Final key-value classes must be registered with the job.

Differences:

• Palindrome testing takes approximately time for a token of length .
• Iterative factorial takes multiplications and may require BigInteger.
• Armstrong testing uses the number of digits and computes approximately powers.

A reducer becomes useful only if a global summary is required, such as counting all palindromes, finding the number of Armstrong values, or multiplying partial results.

A reliable driver should perform the following tasks:

• Validate that input and output arguments are supplied.
• Select the required mapper and reducer classes.
• Register intermediate and final key-value classes correctly.
• Set zero reducers for independent record-classification jobs.
• Configure input and output paths.
• Return a meaningful process exit code.
• Ensure that the output directory does not already exist, or deliberately remove it when appropriate.

Type rules:

• Mapper output types must match setMapOutputKeyClass() and setMapOutputValueClass().
• Reducer input types must match mapper output types.
• Final types must match setOutputKeyClass() and setOutputValueClass().

Validation practices:

• Catch NumberFormatException for malformed values.
• Reject negative values where the operation is undefined.
• Use counters such as INVALID_NUMBER to monitor data quality.
• Use LongWritable, Text, or custom Writable types according to the data range.
• Use BigInteger internally for factorial or power computations that can overflow primitive types.

Testing should include normal input, empty lines, extra spaces, duplicate values, negative numbers, zero, malformed tokens, large values, and an empty input file.