l OM oARc PSD|29 85 3 65

UNIT IV NOTES
SYLLABUS:

UNIT IV MAP REDUCE APPLICATIONS

MapReduce workflows – unit tests with MRUnit – test data and local tests – anatomy of
MapReduce job run – classic Map-reduce – YARN – failures in classic Map-reduce and YARN –
job scheduling – shuffle and sort – task execution – MapReduce types – input formats – output
formats.
 l OM oARc PSD|29 85 3 65

Map-Reduce
 Map-Reduce
A MapReduce is a data processing tool which is used to process the data parallelly in a distributed
form. It was developed in 2004, on the basis of paper titled as "MapReduce: Simplified Data
Processing on Large Clusters," published by Google. MapReduce is a software framework and
programming model used for processing huge amounts of data. The MapReduce is a paradigm
which has two phases, the mapper phase, and the reducer phase. In the Mapper, the input is given
in the form of a key-value pair. The output of the Mapper is fed to the reducer as input. The reducer
runs only after the Mapper is over. The reducer too takes input in key-value format, and the output
of reducer is the final output.

 Different phases of Map-Reduce

The data goes through the following phases of MapReduce in Big Data

Input Splits:

An input to a MapReduce in Big Data job is divided into fixed-size pieces called input splits
Input split is a chunk of the input that is consumed by a single map

Mapping

This is the very first phase in the execution of map-reduce program. In this phase data in each
split is passed to a mapping function to produce output values. In our example, a job of mapping
phase is to count a number of occurrences of each word from input splits (more details about
input-split is given below) and prepare a list in the form of <word, frequency>

Shuffling

This phase consumes the output of Mapping phase. Its task is to consolidate the relevant records
from Mapping phase output. In our example, the same words are clubed together along with
their respective frequency.

Reducing

In this phase, output values from the Shuffling phase are aggregated. This phase combines
values from Shuffling phase and returns a single output value. In short, this phase summarizes
the complete dataset.

 How Map Reduce works

‘ Page 1
 l OM oARc PSD|29 85 3 65

How Hadoop MapReduce works by understanding the end-to-end Hadoop MapReduce job
execution flow with components in detail.

Input Files:-
The data for a MapReduce task is stored in input files, and input files typically lives in HDFS.
The format of these files is arbitrary, while line-based log files and binary format can also be
used.

Input Format:-
Now, Input Format defines how these input files are split and read. It selects the files or other
objects that are used for input. Input Format creates Input Split.

Input Splits:-
It is created by Input Format, logically represent the data which will be processed by an
individual Mapper. One map task is created for each split; thus, the number of map tasks will be
equal to the number of Input Splits. The split is divided into records and each record will be
processed by the mapper.

Record Reader:-
It communicates with the Input Split in Hadoop MapReduce and converts the data into key-
value pairs suitable for reading by the mapper. By default, it uses Text Input Format for
converting data into a key-value pair. Record Reader communicates with the Input Split until the
file reading is not completed. It assigns byte offset (unique number) to each line present in the
file. Further, these key-value pairs are sent to the mapper for further processing.

Mapper:-
It processes each input record (from Record Reader) and generates new key-value pair, and this
key-value pair generated by Mapper is completely different from the input pair. The output of
Mapper is also known as intermediate output which is written to the local disk. The output of the
Mapper is not stored on HDFS as this is temporary data and writing on HDFS will create
unnecessary copies (also HDFS is a high latency system). Mappers output is passed to the
combiner for further process.

Combiner:-
The combiner is also known as ‘Mini-reducer’. Hadoop MapReduce Combiner performs local
aggregation on the mappers’ output, which helps to minimize the data transfer between mapper
and reducer (we will see reducer below). Once the combiner functionality is executed, the
output is then passed to the partitioner for further work.
‘ Page 2
 l OM oARc PSD|29 85 3 65

Partitioner:-
Hadoop MapReduce, Partitioner comes into the picture if we are working on more than one
reducer (for one reducer partitioner is not used). Partitioner takes the output from combiners and
performs partitioning. Partitioning of output takes place on the basis of the key and then sorted.
By hash function, key (or a subset of the key) is used to derive the partition. According to the
key value in MapReduce, each combiner output is partitioned, and a record having the same key
value goes into the same partition, and then each partition is sent to a reducer. Partitioning
allows even distribution of the map output over the reducer.

Shuffling and Sorting:-

Now, the output is Shuffled to the reduce node (which is a normal slave node but reduce phase
will run here hence called as reducer node). The shuffling is the physical movement of the data
which is done over the network. Once all the mappers are finished and their output is shuffled on
the reducer nodes, then this intermediate output is merged and sorted, which is then provided as
input to reduce phase.

Reducer:-
It takes the set of intermediate key-value pairs produced by the mappers as the input and then
runs a reducer function on each of them to generate the output. The output of the reducer is the
final output, which is stored in HDFS.

Record Writer:-
It writes these output key-value pair from the Reducer phase to the output files.

Output Format:-
The way these output key-value pairs are written in output files by Record Writer is determined
by the Output Format. Output Format instances provided by the Hadoop are used to write files in
HDFS or on the local disk. Thus, the final output of reducer is written on HDFS by Output
Format instances.

Let us try to understand the two tasks Map &f Reduce with the help of a small diagram.

 Developing a Map Reduce Application

Configuration API:-
A Configuration class is used to access the configuration XML and can be combined (if a var is
repeated, last is used). Variables can also be expanded using system properties.
‘ Page 3
 l OM oARc PSD|29 85 3 65

Configuring the Development Environment:-

All JAR's from top level Hadoop directory must be added to the IDE. Also, you can have local
and cluster file configurations.

Generic Options Parser, Tool and Tool Runner:-

 Generic Options Parser interprets Hadoop command-line options and sets them on a
Configuration object
 Tool is an interface to use the above class

Writing Unit Tests:-

Mapper Unit Test
Because Mapper and Reducers writes to Context files (instead of returning the result) a mock for
the Context object is needed.
Reducer Unit Test
Reducer unit test is similar to mapper unit test .

Running locally and in a cluster on Test Data:-

Locally Using the Tool interface you could write a driver to configure the local job.
Cluster No code changes are needed, just to pack the Jar.

The MapReduce Web UI:-

It consists of following information
1. Hadoop installation: version, compilation, jobtracker state...
2. Summary of the cluster: capacity, utilization, mr running, jobs, tasktrackers, slots,
blacklisted Tasktrackers
3. Job Scheduler: Running and failed jobs with id's, owner, name...
4. Link to Jobtracker Logs: historic

Hadoop Logs:-
Logfiles can be found on the local fs of each TaskTracker and if JVM reuse is enabled, each log
accumulates the entire JVM run. Anything written to standard output or error is directed to the
relevant logfile.

Tuning a Job to improve performance: -

Hadoop performance tuning helps in optimizing Hadoop cluster performance and achieve the
best results while running MapReduce jobs. It is very important for the Hadoop administrators to
be familiar with the several hardware specifications such as RAM capacity, the number of disks
‘ Page 4
 l OM oARc PSD|29 85 3 65

mounted on the DataNodes, number of CPU cores, the number of physical or virtual cores, NIC
Cards, etc. We can choose the performance tuning tips and tricks on the basis of the amount of
data to be moved and on the type of Hadoop job to be run in production. The best and the most
effective performance tuning helps in achieving maximum performance.

 Unit tests with MR unit

Testing a Hadoop job requires a lot of effort not related to the job. You must configure it to run
locally, create a sample input file, run the job on your sample input, and then compare to an
expected output file. This not only takes time, but makes your tests run very slow due to all the
file I/O. A unit test library designed to facilitate easy integration between your MapReduce
development process and standard development and testing tools such as Junit. With MRUnit,
there are no test files to create, no configuration parameters to change, and generally less test
code. You can cut the clutter and focus on the meat of your tests.

MRUnit is a JUnit-based Java library that allows us to unit test Hadoop MapReduce programs.
This makes it easy to develop as well as to maintain Hadoop MapReduce code bases. MRUnit
supports testing Mappers and Reducers separately as well as testing MapReduce computations as
a whole.

MRTest detects a design fault because in some configuration the program not always calculates
the average temperature rightly. In this case, the program is wrongly designed because the
Combiner functionality that calculates the average is not commutive/associative. Depending on
how Hadoop executes the program, sometimes it could obtain the right outcome and other fails.
MRTest executes automatically the unit test case under a representative subset of configurations
and compares their outcomes.

 Test data and local tests in MapReduce

 How does Hadoop execute a MapReduce program

MapReduce can be used to work with a solitary method call: submit() on a Job object (you can
likewise call waitForCompletion(), which presents the activity on the off chance that it hasn’t
been submitted effectively, at that point sits tight for it to finish).

1. Client: Submitting the MapReduce job.

2. Yarn node manager: In a cluster, it monitors and launches the compute containers on
machines.
3. Yarn resource manager: Handles the allocation of computing resources coordination on
the cluster.
‘ Page 5
 l OM oARc PSD|29 85 3 65

4. MapReduce application master Facilitates the tasks running the MapReduce work.
5. Distributed Filesystem: Shares job files with other entities.

 Anatomy of a Map Reduce Job Run

A job run in classic MapReduce is illustrated in below figure. At the highest level, there are
four independent entities:
• The client, which submits the MapReduce job.
• The jobtracker, which coordinates the job run. The jobtracker is a Java application whose
main class is JobTracker.
• The tasktrackers, which run the tasks that the job has been split into. Tasktrackers are Java
applications whose main class is TaskTracker.
• The distributed filesystem, which is used for sharing job files between the other entities.

Job Submission:
The submit() method on Job creates an internal JobSummitter instance and calls submitJobInternal() on it
(step 1 in figure). The job submission process implemented by JobSummitter does the following:
 Asks the jobtracker for a new job ID (by calling getNewJobId() on JobTracker) (step 2).
 Computes the input splits for the job. Copies the resources needed to run the job, including the job JAR
file, the configuration file, and the computed input splits, to the
jobtracker’s filesystem in a directory named after the job ID. (step 3).
 Tells the jobtracker that the job is ready for execution (by calling submitJob() on) (step 4).
sharing job files between the other entities.

Job Initialization:
When the JobTracker receives a call to its submitJob() method, it puts it into an internal queue from
where the job scheduler will pick it up and initialize it. Initialization involves an object to represent the
job being run (step 5). To create the list of tasks to run, the job scheduler first retrieves the input splits
computed by the client from the shared filesystem (step 6). It then creates one map task for each split.

Task Assignment:
Tasktrackers run a simple loop that periodically sends heartbeat method calls to the jobtracker.
Heartbeats tell the jobtracker that a tasktracker is alive As a part of the heartbeat, a tasktracker will indicate
whether it is ready to run a new task, and if it is, the jobtracker will allocate it a task, which it
communicates to the tasktracker using the heartbeat return value (step 7).

Task Execution:
Now that the tasktracker has been assigned a task, the next step is for it to run the task. First, it localizes
the job JAR by copying it from the shared filesystem to the tasktracker’s filesystem. It also copies any files
needed from the distributed cache by the application to the local disk; see “Distributed Cache” on page 288
(step 8). TaskRunner launches a new Java Virtual Machine (step 9) to run each task in (step 10).

Progress and Status Updates:

MapReduce jobs are long-running batch jobs, taking anything from minutes to hours to run. Because this
is a significant length of time, it’s important for the user to get feedback on how the job is progressing. A
job and each of its tasks have a status. When a task is running, it keeps track of its progress, that is, the
proportion of the task completed.

Job Completion:
When the jobtracker receives a notification that the last task for a job is complete (this will be the special
job cleanup task), it changes the status for the job to “successful.”

‘ Page 6
 l OM oARc PSD|29 85 3 65

 Scalability shortcomings of classic MapReduce is overcome by YARN

In MapReduce framework, MapReduce job (MapReduce application) is divided between number
of tasks called mappers and reducers. Each task runs on one of the machines (DataNode) of the
cluster, and each machine has a limited number of predefined slots (map slot, reduce slot) for
running tasks concurrently. Here, JobTracker is responsible for both managing the cluster’s
resources and driving the execution of the MapReduce job. It reserves and schedules slots for all
tasks, configures, runs and monitors each task, and if a task fails, it allocates a new slot and
reattempts the task. After a task finishes, the job tracker cleans up temporary resources and releases
the task’s slot to make it available for other jobs.

Problems with this approach in Hadoop 1.0:

1. It limits scalability: JobTracker runs on single machine doing several task like
 Resource management
 Job and task scheduling and
 Monitoring

Although there are so many machines (DataNode) available; they are not getting
used. This limits scalability.

2. Availability Issue: In Hadoop 1.0, JobTracker is single Point of availability. This

means if JobTracker fails, all jobs must restart.
3. Problem with Resource Utilization: In Hadoop 1.0, there is concept of predefined
number of map slots and reduce slots for each TaskTrackers. Resource Utilization
issues occur because maps slots might be ‘full’ while reduce slots is empty (and
vice-versa). Here the compute resources (DataNode) could sit idle which are
reserved for Reduce slots even when there is immediate need for those resources
to be used as Mapper slots.
4. Limitation in running non-MapReduce Application: In Hadoop 1.0, Job tracker
was tightly integrated with MapReduce and only supporting application that obeys
MapReduce programming framework can run on Hadoop. Hadoop distributed file

‘ Page 7
 l OM oARc PSD|29 85 3 65

system (HDFS) makes it cheap to store large amounts of data, and its scalable
MapReduce analysis engine makes it possible to extract insights from that data.
MapReduce works on batch-driven data analysis, where the input data is
partitioned into smaller batches that can be processed in parallel across many
machines in the Hadoop cluster. But MapReduce, while powerful enough toexpress
many data analysis algorithms, is not always the optimal choice of programming
paradigm. It ‘s often desirable to run other computation paradigms in the Hadoop
cluster here are some examples.
 Problem in performing real-time analysis: MapReduce is batch
driven. What if I want to do perform real time analysis instead of
batch-processing (where results is available after several hours).
There are many applications which need results in real time like
fraud detection algorithm. There are real time engines like Apache
Storm which can work better in this case. But in Hadoop 1.0, due to
tight coupling these engines cannot run independently.
 Problem in running Message-Passing approach: It is a stateful
process that runs on each node of a distributed network. The
processes communicate with each other by sending messages, and
alter their state based on the messages they receive. This is not
possible in MapReduce.
 Problem in running Ad-hoc query: Many users like to query their
big data using SQL. Apache Hive can execute a SQL query as a
series of MapReduce jobs, but it has shortcomings in terms of
performance. Recently, some new approaches such as Apache Tajo
, Facebook’s Presto and Cloudera’s Impala drastically improve the
performance, but they require to run services in other form than
MapReduce form. It is not possible to run all such non Map Reduce
jobs on Hadoop Cluster. Such jobs have to “disguise” themselves as
mappers and reducers in order to be able to run on Hadoop 1.0.

 Explain the anatomy a job run in YARN (MapReduce 2.0)

We can run a MapReduce job with a single method call: submit () on a Job object (you can also call
waitForCompletion (), which submits the job if it hasn’t been submitted already, then waits for it to finish).
The whole process is illustrated in figure 7. At the highest level, there are five independent entities:

• The client, which submits the MapReduce job.

• The YARN resource manager, which coordinates the allocation of compute resources on the cluster.
• The YARN node managers, which launch and monitor the compute containers on machines in the cluster.
• The MapReduce application master, which coordinates the tasks running the MapReduce job. The
application master and the MapReduce tasks run in containers that are scheduled by the resource manager
and managed by the node managers.

Job Submission
The submit() method on Job creates an internal JobSubmitter instance and calls submitJobInternal() on it
(step 1 in figure ). Having submitted the job, waitForCompletion() polls the job’s progress once per second
and reports the progress to the console if it has changed since the last report. When the job completes
successfully, the job counters are displayed. Otherwise, the error that caused the job to fail is logged to the
console.

Job Initialization

‘ Page 8
 l OM oARc PSD|29 85 3 65

When the resource manager receives a call to its submitApplication() method, it hands off the request to
the YARN scheduler. The scheduler allocates a container, and the resource manager then launches the
application master’s process there, under the node manager’s management (steps 5a and 5b).

Task Assignment
If the job does not qualify for running as an uber task, then the application master requests containers for
all the map and reduce tasks in the job from the resource manager (step 8). Requests for map tasks are made
first and with a higher priority than those for reduce tasks, since all the map tasks must complete before the
sort phase of the reduce can start. Requests for reduce tasks are not made until 5% of map tasks have
completed.

Task Execution
Once a task has been assigned resources for a container on a particular node by the resource manager’s
scheduler, the application master starts the container by contacting the node manager (steps 9a and 9b).
The task is executed by a Java application whose main class is YarnChild. Before it can run the task, it
localizes the resources that the task needs, including the job configuration and JAR file, and any files from
the dis‐ tributed cache (step 10). Finally, it runs the map or reduce task (step 11).

Job Completion
When the application master receives a notification that the last task for a job is com‐ plete, it changes the
status for the job to “successful.” Then, when the Job polls for status, it learns that the job has completed
successfully, so it prints a message to tell the user and then returns from the waitForCompletion() method.
Job statistics and counters are printed to the console at this point. Finally, on job completion, the application
master and the task containers clean up their working state.

 Various failure in classic MapReduce

 Failures

In the real world, user code is buggy, processes crash, and machines fail. One of the major benefits of
using Hadoop is its ability to handle such failures and allow your job to complete.

‘ Page 9
 l OM oARc PSD|29 85 3 65

 Failures in Classic MapReduce

In the MapReduce 1 runtime there are three failure modes to consider: failure of the running task, failure
of the tastracker, and failure of the jobtracker.

 Task Failure

Consider first the case of the child task failing. The most common way that this happens is when user code
in the map or reduce task throws a runtime exception. If this happens, the child JVM reports the error back
to its parent tasktracker, before it exits. The error ultimately makes it into the user logs. The tasktracker
marks the task attempt as failed, freeing up a slot to run another task.

When the jobtracker is notified of a task attempt that has failed (by the tasktracker’s heartbeat call), it will
reschedule execution of the task. The jobtracker will try to avoid rescheduling the task on a tasktracker
where it has previously failed. Furthermore, if a task fails four times (or more), it will not be retried further.

A task attempt may also be killed, which is different from it failing. A task attempt may be killed because
it is a speculative duplicate, or because the tasktracker it was running on failed, and the jobtracker marked
all the task attempts running on it as killed. Users may also kill or fail task attempts using the web UI or
the command line (type hadoop job to see the options). Jobs may also be killed by the same mechanisms.

 Tasktracker Failure

Failure of a tasktracker is another failure mode. If a tasktracker fails by crashing, or running very slowly,
it will stop sending heartbeats to the jobtracker (or send them very infrequently). The jobtracker will notice
a tasktracker that has stopped sending heartbeats (if it hasn’t received one for 10 minutes, configured via
the [Link] [Link] property, in milliseconds) and remove it from its pool of
tasktrackers to schedule tasks on. The jobtracker arranges for map tasks that were run and completed
successfully on that tasktracker to be rerun if they belong to incomplete jobs, since their intermediate output
residing on the failed tasktracker’s local filesystem may not be accessible to the reduce task. Any tasks in
progress are also rescheduled.

 Jobtracker Failure

Failure of the jobtracker is the most serious failure mode. Hadoop has no mechanism for dealing with
failure of the jobtracker. it is a single point of failure, so in this case the job fails. However, this failure
mode has a low chance of occurring, since the chance of a particular machine failing is low. The good news
is that the situation is improved in YARN, since one of its design goals is to eliminate single points of
failure in MapReduce.

After restarting a jobtracker, any jobs that were running at the time it was stopped will need to be re-
submitted. There is a configuration option that attempts to recover any running jobs, however it is known
not to work reliably, so should not be used.

 Various failure in YARN

 Failures in YARN

For MapReduce programs running on YARN, we need to consider the failure of any of the following
entities: the task, the application master, the node manager, and the resource manager.

 Task Failure

Failure of the running task is similar to the classic case. Runtime exceptions and sudden exits of the JVM
are propagated back to the application master and the task attempt is marked as failed. Likewise, hanging

‘ Page 10
 l OM oARc PSD|29 85 3 65

tasks are noticed by the application master by the absence of a ping over the umbilical channel (the
timeout is set by [Link] out), and again the task attempt is marked as failed.

The configuration properties for determining when a task is considered to be failed are the same as the
classic case: a task is marked as failed after four attempts (set by [Link] for map
tasks and [Link] for reducer tasks). A job will be failed if more than
[Link] percent of the map tasks in the job fail, or more than
[Link] percent of the reduce tasks fail.

 Application Master Failure

Just like MapReduce tasks are given several attempts to succeed (in the face of hardware or network
failures) applications in YARN are tried multiple times in the event of failure. By default, applications
are marked as failed if they fail once.

An application master sends periodic heartbeats to the resource manager, and in the event of application
master failure, the resource manager will detect the failure and start a new instance of the master running
in a new container (managed by a node manager). In the case of the MapReduce application master, it
can recover the state of the tasks that had already been run by the (failed) application so they don’t have
to be rerun. By default, recovery is not enabled, so failed application masters will not rerun all their tasks,
but you can turn it on by setting [Link] [Link] to true.

 Node Manager Failure

If a node manager fails, then it will stop sending heartbeats to the resource manager, and the node
manager will be removed from the resource manager’s pool of available nodes. The property
[Link]-intervalms, which defaults to 600000 (10 minutes),
determines the minimum time the resource manager waits before considering a node manager that has
sent no heartbeat in that time as failed.

Any task or application master running on the failed node manager will be recovered using the
mechanisms described in the previous two sections. Node managers may be blacklisted if the number of
failures for the application is high. Blacklisting is done by the application master, and for MapReduce the
application master will try to reschedule tasks on different nodes if more than three tasks fail on a node
manager.

 Resource Manager Failure

Failure of the resource manager is serious, since without it neither jobs nor task containers can be launched.
The resource manager was designed from the outset to be able to recover from crashes, by using a
checkpointing mechanism to save its state to persistent storage, although at the time of writing the latest
release did not have a complete implementation. After a crash, a new resource manager instance is brought
up (by an adminstrator) and it recovers from the saved state. The state consists of the node managers in the
system as well as the running applications.

 The type of scheduler in MapReduce

There are mainly 3 types of Schedulers in Hadoop:

1. FIFO (First in First Out) Scheduler.

2. Capacity Scheduler.
3. Fair Scheduler.

‘ Page 11
 l OM oARc PSD|29 85 3 65

These Schedulers are actually a kind of algorithm that we use to schedule tasks in a Hadoop
cluster when we receive requests from different-different clients.

A Job queue is nothing but the collection of various tasks that we have received from our
various clients. The tasks are available in the queue and we need to schedule this task on the
basis of our requirements.

1. FIFO Scheduler
As the name suggests FIFO i.e. First In First Out, so the tasks or application that comes first will
be served first. This is the default Scheduler we use in Hadoop. The tasks are placed in a queue
and the tasks are performed in their submission order. In this method, once the job is scheduled,
no intervention is allowed. So sometimes the high-priority process has to wait for a long time since
the priority of the task does not matter in this method.
Advantage:
 No need for configuration
 First Come First Serve
 simple to execute
Disadvantage:
 Priority of task doesn’t matter, so high priority jobs need to wait
 Not suitable for shared cluster

2. Capacity Scheduler
In Capacity Scheduler we have multiple job queues for scheduling our tasks. The Capacity
Scheduler allows multiple occupants to share a large size Hadoop cluster. In Capacity Scheduler
corresponding for each job queue, we provide some slots or cluster resources for performing job
operation. Each job queue has it’s own slots to perform its task. In case we have tasks to perform
in only one queue then the tasks of that queue can access the slots of other queues also as they are
free to use, and when the new task enters to some other queue then jobs in running in its own slots
of the cluster are replaced with its own job.

Capacity Scheduler also provides a level of abstraction to know which occupant is utilizing the
more cluster resource or slots, so that the single user or application doesn’t take dis-appropriate or
unnecessary slots in the cluster. The capacity Scheduler mainly contains 3 types of the queue that
are root, parent, and leaf which are used to represent cluster, organization, or any subgroup,
application submission respectively.

Advantage:
 Best for working with Multiple clients or priority jobs in a Hadoop cluster
 Maximizes throughput in the Hadoop cluster

‘ Page 12
 l OM oARc PSD|29 85 3 65

Disadvantage:
 More complex
 Not easy to configure for everyone

3. Fair Scheduler
The Fair Scheduler is very much similar to that of the capacity scheduler. The priority of the job
is kept in consideration. With the help of Fair Scheduler, the YARN applications can share the
resources in the large Hadoop Cluster and these resources are maintained dynamically so no need
for prior capacity. The resources are distributed in such a manner that all applications within a
cluster get an equal amount of time. Fair Scheduler takes Scheduling decisions on the basis of
memory, we can configure it to work with CPU also.

As we told you it is similar to Capacity Scheduler but the major thing to notice is that in Fair
Scheduler whenever any high priority job arises in the same queue, the task is processed in parallel
by replacing some portion from the already dedicated slots.

Advantages:
 Resources assigned to each application depend upon its priority.
 it can limit the concurrent running task in a particular pool or queue.
Disadvantages: The configuration is required.

 The shuffle and sort in Hadoop MapReduce

Shuffling is the process by which it transfers mappers intermediate output to the reducer.
Reducer gets 1 or more keys and associated values on the basis of reducers.

The intermediated key value generated by mapper is sorted automatically by key. In Sort phase
merging and sorting of map output takes place.

Shuffling and Sorting in Hadoop occurs simultaneously.

‘ Page 13
 l OM oARc PSD|29 85 3 65

Shuffling in MapReduce

The process of transferring data from the mappers to reducers is shuffling. It is also the process by
which the system performs the sort. Then it transfers the map output to the reducer as input. This
is the reason shuffle phase is necessary for the reducers. Otherwise, they would not have anyinput
(or input from every mapper). Since shuffling can start even before the map phase has finished.
So this saves some time and completes the tasks in lesser time.

Sorting in MapReduce

MapReduce Framework automatically sort the keys generated by the mapper. Thus, before starting
of reducer, all intermediate key-value pairs get sorted by key and not by value. It does not sort
values passed to each reducer. They can be in any order. Sorting in a MapReduce job helps reducer
to easily distinguish when a new reduce task should start.

This saves time for the reducer. Reducer in MapReduce starts a new reduce task when the next
key in the sorted input data is different than the previous. Each reduce task takes key value pairs
as input and generates key-value pair as output.

The important thing to note is that shuffling and sorting in Hadoop MapReduce are will not take
place at all if you specify zero reducers (setNumReduceTasks(0)). If reducer is zero, then the
MapReduce job stops at the map phase. And the map phase does not include any kind of sorting
(even the map phase is faster).

Secondary Sorting in MapReduce

If we want to sort reducer values, then we use a secondary sorting technique. This technique
enables us to sort the values (in ascending or descending order) passed to each reducer.

 The different types of input formats in Hadoop MapReduce

Hadoop Input Format describes the input-specification for execution of the Map-Reduce job.
InputFormat describes how to split up and read input files. In MapReduce job execution,
InputFormat is the first step. It is also responsible for creating the input splits and dividing them
into records. Input files store the data for MapReduce job. Input files reside in HDFS. Although
these files format is arbitrary, we can also use line-based log files and binary format.

Types of InputFormat in MapReduce

There are different types of MapReduce InputFormat in Hadoop which are used for different
purpose. Let’s discuss the Hadoop InputFormat types below:

‘ Page 14
 l OM oARc PSD|29 85 3 65

1. FileInputFormat
It is the base class for all file-based InputFormats. FileInputFormat also specifies input directory
which has data files location. When we start a MapReduce job execution, FileInputFormat
provides a path containing files to read. This InpuFormat will read all files. Then it divides these
files into one or more InputSplits.
2. TextInputFormat
It is the default InputFormat. This InputFormat treats each line of each input file as a separate
record. It performs no parsing. TextInputFormat is useful for unformatted data or line-based
records like log files. Hence,
 Key – It is the byte offset of the beginning of the line within the file (not whole file one
split). So it will be unique if combined with the file name.
 Value – It is the contents of the line. It excludes line terminators.
3. KeyValueTextInputFormat
It is similar to TextInputFormat. This InputFormat also treats each line of input as a separate
record. While the difference is that TextInputFormat treats entire line as the value, but the
KeyValueTextInputFormat breaks the line itself into key and value by a tab character (‘/t’). Hence,
 Key – Everything up to the tab character.
 Value – It is the remaining part of the line after tab character.
4. NlineInputFormat
It is another form of TextInputFormat where the keys are byte offset of the line. And values are
contents of the line. So, each mapper receives a variable number of lines of input with
TextInputFormat and KeyValueTextInputFormat. The number depends on the size of the split.
Also, depends on the length of the lines. So, if want our mapper to receive a fixed number of lines
of input, then we use NLineInputFormat.
N- It is the number of lines of input that each mapper receives.
By default (N=1), each mapper receives exactly one line of input.
Suppose N=2, then each split contains two lines. So, one mapper receives the first two Key-Value
pairs. Another mapper receives the second two key-value pairs.
5. TextInputFormat – TextInputFormat is the default InputFormat. Each record is a line of input.
The key, a LongWritable, is the byte offset within the file of the beginning of the line. The value
is the contents of the line, excluding any line terminators.
6. KeyValueTextInputFormat – TextInputFormat’s keys, being simply the offsets within the
file, are not normally very useful. It is common for each line in a file to be a key-value pair,
separated by a delimiter such as a tab character. For example, this is the kind of output produced
by TextOut putFormat, Hadoop’s default OutputFormat. To interpret such files correctly,
KeyValueTextInputFormat is appropriate.
7. StreamInputFormat – Hadoop comes with a InputFormat for streaming which can be used
outside streaming and can be used for processing XML documents. You can use it by setting your
input format to StreamInputFormat and setting the [Link] property to
[Link]. The reader is configured by
setting job configuration properties to tell it the patterns for the start and end tags.
8. SequenceFileInputFormat – Hadoop MapReduce is not restricted to processing textual data.
It has support for binary formats, too. Hadoop’s sequence file format stores sequences of binary
key-value pairs. Sequence files are well suited as a format for MapReduce data because they are
splittable (they have sync points so that readers can synchronize with record boundaries from an
arbitrary point in the file, such as the start of a split), they support compression as a part of the
format, and they can store arbitrary types using a variety of serialization frameworks.
9. SequenceFileAsTextInputFormat – SequenceFileAsTextInputFormat is a variant of
SequenceFileInputFormat that converts the sequence file’s keys and values to Text objects. The
conversion is performed by calling toString() on the keys and values. This format makes sequence
files suitable input for Streaming
10. SequenceFileAsBinaryInputFormat – SequenceFileAsBinaryInputFormat is a variant of
SequenceFileInputFormat that retrieves the sequence file’s keys and values as opaque binary
’ Page 15
 l OM oARc PSD|29 85 3 65

objects. They are encapsulated as BytesWritable objects, and the application is free to interpret the
underlying byte array as it pleases.
11. FixedLengthInputFormat – FixedLengthInputFormat is for reading fixed-width binary
records from a file, when the records are not separated by delimiters. The record size must be set
via [Link].
12. Multiple Inputs – Although the input to a MapReduce job may consist of multiple input files
(constructed by a combination of file globs, filters, and plain paths), all of the input is interpreted
by a single InputFormat and a single Mapper. What often happens, however, is that the data format
evolves over time, so you have to write your mapper to cope with all of your legacy formats. Or
you may have data sources that provide the same type of data but in different formats. This arises
in the case of performing joins of different datasets .
14. Database Input – DBInputFormat is an input format for reading data from a relational
database, using JDBC. Because it doesn’t have any sharding capabilities, you need to be careful
not to overwhelm the database from which you are reading by running too many mappers. For
this reason, it is best used for loading relatively small datasets, perhaps for joining with larger
datasets from HDFS using MultipleInputs. The corresponding output format is DBOutputFormat,
which is useful for dumping job outputs (of modest size) into a database.
15. Lazy Output – FileOutputFormat subclasses will create output (part-r-nnnnn) files, even if
they are empty. Some applications prefer that empty files not be created, which is where Lazy
OutputFormat helps. It is a wrapper output format that ensures that the output file is created only
when the first record is emitted for a given partition. To use it, call its setOutputFormatClass()
method with the JobConf and the underlying output format.

 Features of MapReduce

1. Highly scalable

A framework with excellent scalability is Apache Hadoop MapReduce. This is because of its
capacity for distributing and storing large amounts of data across numerous servers. These servers
can all run simultaneously and are all reasonably priced.

By adding servers to the cluster, we can simply grow the amount of storage and computing power.
We may improve the capacity of nodes or add any number of nodes (horizontal scalability) to
attain high computing power. Organizations may execute applications from massive sets of nodes,
potentially using thousands of terabytes of data, thanks to Hadoop MapReduce programming.

2. Versatile

Businesses can use MapReduce programming to access new data sources. It makes it possible for
companies to work with many forms of data. Enterprises can access both organized and
unstructured data with this method and acquire valuable insights from the various data sources.

Since Hadoop is an open-source project, its source code is freely accessible for review, alterations,
and analyses. This enables businesses to alter the code to meet their specific needs. The
MapReduce framework supports data from sources including email, social media, and
clickstreams in different languages.

3. Secure

The MapReduce programming model uses the HBase and HDFS security approaches, and only
authenticated users are permitted to view and manipulate the data. HDFS uses a replication

‘ Page 16
 l OM oARc PSD|29 85 3 65

technique in Hadoop 2 to provide fault tolerance. Depending on the replication factor, it makes a
clone of each block on the various machines. One can therefore access data from the other devices
that house a replica of the same data if any machine in a cluster goes down. Erasure coding has
taken the role of this replication technique in Hadoop 3. Erasure coding delivers the same level of
fault tolerance with less area. The storage overhead with erasure coding is less than 50%.

4. Affordability

With the help of the MapReduce programming framework and Hadoop’s scalable design, big data
volumes may be stored and processed very affordably. Such a system is particularly cost-effective
and highly scalable, making it ideal for business models that must store data that is constantly
expanding to meet the demands of the present.

In terms of scalability, processing data with older, conventional relational database management
systems was not as simple as it is with the Hadoop system. In these situations, the company had
to minimize the data and execute classification based on presumptions about how specific data
could be relevant to the organization, hence deleting the raw data. The MapReduce programming
model in the Hadoop scale-out architecture helps in this situation.

5. Fast-paced

The Hadoop Distributed File System, a distributed storage technique used by MapReduce, is a
mapping system for finding data in a cluster. The data processing technologies, such as
MapReduce programming, are typically placed on the same servers that enable quicker data
processing.

Thanks to Hadoop’s distributed data storage, users may process data in a distributed manner across
a cluster of nodes. As a result, it gives the Hadoop architecture the capacity to process data
exceptionally quickly. Hadoop MapReduce can process unstructured or semi-structured data in
high numbers in a shorter time.

6. Based on a simple programming model

Hadoop MapReduce is built on a straightforward programming model and is one of the

technology’s many noteworthy features. This enables programmers to create MapReduce
applications that can handle tasks quickly and effectively. Java is a very well-liked and simple-to-
learn programming language used to develop the MapReduce programming model.

Java programming is simple to learn, and anyone can create a data processing model that works
for their company. Hadoop is straightforward to utilize because customers don’t need to worry
about computing distribution. The framework itself does the processing.

7. Parallel processing-compatible

The parallel processing involved in MapReduce programming is one of its key components. The
tasks are divided in the programming paradigm to enable the simultaneous execution of
independent activities. As a result, the program runs faster because of the parallel processing,
which makes it simpler for the processes to handle each job. Multiple processors can carry out
these broken-down tasks thanks to parallel processing. Consequently, the entire software runs
faster.

‘ Page 17
 l OM oARc PSD|29 85 3 65

8. Reliable

The same set of data is transferred to some other nodes in a cluster each time a collection of
information is sent to a single node. Therefore, even if one node fails, backup copies are always
available on other nodes that may still be retrieved whenever necessary. This ensures high data
availability.

The framework offers a way to guarantee data trustworthiness through the use of Block Scanner,
Volume Scanner, Disk Checker, and Directory Scanner modules. Your data is safely saved in the
cluster and is accessible from another machine that has a copy of the data if your device fails or
the data becomes corrupt.

9. Highly available

Hadoop’s fault tolerance feature ensures that even if one of the DataNodes fails, the user may still
access the data from other DataNodes that have copies of it. Moreover, the high accessibility
Hadoop cluster comprises two or more active and passive NameNodes running on hot standby.
The active NameNode is the active node. A passive node is a backup node that applies changes
made in active NameNode’s edit logs to its namespace.

10. Flexibility

MapReduce programming enables companies to access new sources of data. It enables companies
to operate on different types of data. It allows enterprises to access structured as well as
unstructured data, and derive significant value by gaining insights from the multiple sources of
data. Additionally, the MapReduce framework also provides support for the multiple languages
and data from sources ranging from email, social media, to clickstream. The MapReduce processes
data in simple key-value pairs thus supports data type including meta-data, images, and large files.
Hence, MapReduce is flexible to deal with data rather than traditional DBMS.

11. Parallel Programming

One of the major aspects of the working of MapReduce programming is its parallel processing. It
divides the tasks in a manner that allows their execution in parallel. The
parallel processing allows multiple processors to execute these divided tasks. So the entire
program is run in less time.

12. Counters

They act pseudo checkpoints for your program, with their values getting incremented when a
certain event is triggered. Counters are counter-part of logs, but in numeric format. The counter
values can be used to provide statistical information about the health of the node and, in turn about
the progress of the job. Hadoop provides various build in counters, divided into groups. Each group
either contains a task counter or a job counter.

13. Side Data Distribution

Side data as the name suggests, is the extra data used along side with the main dataset, to aid in
the processing of the main dataset. This side data needs to be present at every mapper and/or
reducer node (which might need side data) in the most optimised manner (keeping in mind the
load on the nodes and the network).

‘ Page 18
 l OM oARc PSD|29 85 3 65

14. Joins

Let us re-use the old dummy dataset of names. Along with this dataset, let us assume another,
equally large dataset of <First_Name, Age>.

Now, suppose you wish to determine how many people with Last_Name starting with P, are aged
25. We have to combine the information from the two datasets, to be able to query it. This is where
the concept of Joins(literally joining 2 or more datasets) comes into picture. Joining in MapReduce
can be performed either by a Mapper, or by a Reducer. If the join is performed by the Mapper, it
is called Map-Side Join. The join performed by the reducer is called the Reduce-Side Join.

15. Sorting

Sorting algorithms form the heart of MapReduce paradigm. Even if your dataset is not sort
oriented, it can, and usually does, utilize the sorting ability of MapRed to sort things out!(faster).

 Speculative Execution

In this Big data Hadoop tutorial, we are going to learn Hadoop speculative execution. Apache Hadoop
does not fix or diagnose slow-running tasks. Instead, it tries to detect when a task is running slower than
expected and launches another, an equivalent task as a backup (the backup task is called as speculative
task). This process is called speculative execution in Hadoop.

In Hadoop, MapReduce breaks jobs into tasks and these tasks run parallel rather than sequential, thus
reduces overall execution time. This model of execution is sensitive to slow tasks (even if they are few in
numbers) as they slow down the overall execution of a job. There may be various reasons for the slowdown
of tasks, including hardware degradation or software misconfiguration, but it may be difficult to detect
causes since the tasks still complete successfully, although more time is taken than the expected time.
Hadoop doesn’t try to diagnose and fix slow running tasks, instead, it tries to detect them and runs backup
tasks for them. This is called speculative execution in Hadoop. These backup tasks are called Speculative
tasks in Hadoop.

 Task JVM Reuse

 Skipping Bad Records

Hadoop provides an optional mode of execution in which the bad records are detected and skipped
in further attempts. This feature can be used when map/reduce tasks crashes deterministically on
certain input. This happens due to bugs in the map/reduce function. The usual course would be to
fix these bugs. But sometimes this is not possible; perhaps the bug is in third party libraries for
which the source code is not available. Due to this, the task never reaches to completion even with
multiple attempts and complete data for that task is lost.

‘ Page 19