Data Flow
First, some terminology. A MapReduce job is a unit of work that the client wants to be
performed: it consists of the input data, the MapReduce program, and configuration
information. Hadoop runs the job by dividing it into tasks, of which there are two types:
map tasks and reduce tasks.
There are two types of nodes that control the job execution process: a jobtracker and
a number of tasktrackers. The jobtracker coordinates all the jobs run on the system by
scheduling tasks to run on tasktrackers. Tasktrackers run tasks and send progress
reports to the jobtracker, which keeps a record of the overall progress of each job. If a
task fails, the jobtracker can reschedule it on a different tasktracker.
Hadoop divides the input to a MapReduce job into fixed-size pieces called input
splits, or just splits. Hadoop creates one map task for each split, which runs the userdefined
map function for each record in the split.
Having many splits means the time taken to process each split is small compared to the
time to process the whole input. So if we are processing the splits in parallel, the processing
is better load-balanced when the splits are small, since a faster machine will be
able to process proportionally more splits over the course of the job than a slower
machine. Even if the machines are identical, failed processes or other jobs running
concurrently make load balancing desirable, and the quality of the load balancing increases
as the splits become more fine-grained.
On the other hand, if splits are too small, the overhead of managing the splits and of
map task creation begins to dominate the total job execution time. For most jobs, a
good split size tends to be the size of an HDFS block, 64 MB by default, although this
can be changed for the cluster (for all newly created files) or specified when each file is
created.
Hadoop does its best to run the map task on a node where the input data resides in
HDFS. This is called the data locality optimization because it doesn’t use valuable cluster
bandwidth. Sometimes, however, all three nodes hosting the HDFS block replicas
for a map task’s input split are running other map tasks, so the job scheduler will look
for a free map slot on a node in the same rack as one of the blocks. Very occasionally
even this is not possible, so an off-rack node is used, which results in an inter-rack
network transfer. The three possibilities are illustrated in Figure 2-2.
Figure 2-2. Data-local (a), rack-local (b), and off-rack (c) map tasks
It should now be clear why the optimal split size is the same as the block size: it is the
largest size of input that can be guaranteed to be stored on a single node. If the split
spanned two blocks, it would be unlikely that any HDFS node stored both blocks, so
some of the split would have to be transferred across the network to the node running
the map task, which is clearly less efficient than running the whole map task using local
data.
Map tasks write their output to the local disk, not to HDFS. Why is this? Map output
is intermediate output: it’s processed by reduce tasks to produce the final output, and
once the job is complete, the map output can be thrown away. So storing it in HDFS
with replication would be overkill. If the node running the map task fails before the
map output has been consumed by the reduce task, then Hadoop will automatically
rerun the map task on another node to re-create the map output.
Reduce tasks don’t have the advantage of data locality; the input to a single reduce task
is normally the output from all mappers. In the present example, we have a single reduce
task that is fed by all of the map tasks. Therefore, the sorted map outputs have to be
transferred across the network to the node where the reduce task is running, where
they are merged and then passed to the user-defined reduce function. The output of
the reduce is normally stored in HDFS for reliability. As explained in Chapter 3, for
each HDFS block of the reduce output, the first replica is stored on the local node, with
other replicas being stored on off-rack nodes. Thus, writing the reduce output does
consume network bandwidth, but only as much as a normal HDFS write pipeline
consumes.
The whole data flow with a single reduce task is illustrated in Figure 2-3. The dotted
boxes indicate nodes, the light arrows show data transfers on a node, and the heavy
arrows show data transfers between nodes.
Figure 2-3. MapReduce data flow with a single reduce task
32 | Chapter 2: MapReduce
The number of reduce tasks is not governed by the size of the input, but instead is
specified independently. In “The Default MapReduce Job” on page 227, you will see
how to choose the number of reduce tasks for a given job.
When there are multiple reducers, the map tasks partition their output, each creating
one partition for each reduce task. There can be many keys (and their associated values)
in each partition, but the records for any given key are all in a single partition. The
partitioning can be controlled by a user-defined partitioning function, but normally the
default partitioner—which buckets keys using a hash function—works very well.
The data flow for the general case of multiple reduce tasks is illustrated in Figure 2-4.
This diagram makes it clear why the data flow between map and reduce tasks is colloquially
known as “the shuffle,” as each reduce task is fed by many map tasks. The
shuffle is more complicated than this diagram suggests, and tuning it can have a big
impact on job execution time, as you will see in “Shuffle and Sort” on page 208.
Figure 2-4. MapReduce data flow with multiple reduce tasks
Finally, it’s also possible to have zero reduce tasks. This can be appropriate when you
don’t need the shuffle because the processing can be carried out entirely in parallel (a
few examples are discussed in “NLineInputFormat” on page 247). In this case, the
only off-node data transfer is when the map tasks write to HDFS (see Figure 2-5).
First, some terminology. A MapReduce job is a unit of work that the client wants to be
performed: it consists of the input data, the MapReduce program, and configuration
information. Hadoop runs the job by dividing it into tasks, of which there are two types:
map tasks and reduce tasks.
There are two types of nodes that control the job execution process: a jobtracker and
a number of tasktrackers. The jobtracker coordinates all the jobs run on the system by
scheduling tasks to run on tasktrackers. Tasktrackers run tasks and send progress
reports to the jobtracker, which keeps a record of the overall progress of each job. If a
task fails, the jobtracker can reschedule it on a different tasktracker.
Hadoop divides the input to a MapReduce job into fixed-size pieces called input
splits, or just splits. Hadoop creates one map task for each split, which runs the userdefined
map function for each record in the split.
Having many splits means the time taken to process each split is small compared to the
time to process the whole input. So if we are processing the splits in parallel, the processing
is better load-balanced when the splits are small, since a faster machine will be
able to process proportionally more splits over the course of the job than a slower
machine. Even if the machines are identical, failed processes or other jobs running
concurrently make load balancing desirable, and the quality of the load balancing increases
as the splits become more fine-grained.
On the other hand, if splits are too small, the overhead of managing the splits and of
map task creation begins to dominate the total job execution time. For most jobs, a
good split size tends to be the size of an HDFS block, 64 MB by default, although this
can be changed for the cluster (for all newly created files) or specified when each file is
created.
Hadoop does its best to run the map task on a node where the input data resides in
HDFS. This is called the data locality optimization because it doesn’t use valuable cluster
bandwidth. Sometimes, however, all three nodes hosting the HDFS block replicas
for a map task’s input split are running other map tasks, so the job scheduler will look
for a free map slot on a node in the same rack as one of the blocks. Very occasionally
even this is not possible, so an off-rack node is used, which results in an inter-rack
network transfer. The three possibilities are illustrated in Figure 2-2.
Figure 2-2. Data-local (a), rack-local (b), and off-rack (c) map tasks
It should now be clear why the optimal split size is the same as the block size: it is the
largest size of input that can be guaranteed to be stored on a single node. If the split
spanned two blocks, it would be unlikely that any HDFS node stored both blocks, so
some of the split would have to be transferred across the network to the node running
the map task, which is clearly less efficient than running the whole map task using local
data.
Map tasks write their output to the local disk, not to HDFS. Why is this? Map output
is intermediate output: it’s processed by reduce tasks to produce the final output, and
once the job is complete, the map output can be thrown away. So storing it in HDFS
with replication would be overkill. If the node running the map task fails before the
map output has been consumed by the reduce task, then Hadoop will automatically
rerun the map task on another node to re-create the map output.
Reduce tasks don’t have the advantage of data locality; the input to a single reduce task
is normally the output from all mappers. In the present example, we have a single reduce
task that is fed by all of the map tasks. Therefore, the sorted map outputs have to be
transferred across the network to the node where the reduce task is running, where
they are merged and then passed to the user-defined reduce function. The output of
the reduce is normally stored in HDFS for reliability. As explained in Chapter 3, for
each HDFS block of the reduce output, the first replica is stored on the local node, with
other replicas being stored on off-rack nodes. Thus, writing the reduce output does
consume network bandwidth, but only as much as a normal HDFS write pipeline
consumes.
The whole data flow with a single reduce task is illustrated in Figure 2-3. The dotted
boxes indicate nodes, the light arrows show data transfers on a node, and the heavy
arrows show data transfers between nodes.
Figure 2-3. MapReduce data flow with a single reduce task
32 | Chapter 2: MapReduce
The number of reduce tasks is not governed by the size of the input, but instead is
specified independently. In “The Default MapReduce Job” on page 227, you will see
how to choose the number of reduce tasks for a given job.
When there are multiple reducers, the map tasks partition their output, each creating
one partition for each reduce task. There can be many keys (and their associated values)
in each partition, but the records for any given key are all in a single partition. The
partitioning can be controlled by a user-defined partitioning function, but normally the
default partitioner—which buckets keys using a hash function—works very well.
The data flow for the general case of multiple reduce tasks is illustrated in Figure 2-4.
This diagram makes it clear why the data flow between map and reduce tasks is colloquially
known as “the shuffle,” as each reduce task is fed by many map tasks. The
shuffle is more complicated than this diagram suggests, and tuning it can have a big
impact on job execution time, as you will see in “Shuffle and Sort” on page 208.
Figure 2-4. MapReduce data flow with multiple reduce tasks
Finally, it’s also possible to have zero reduce tasks. This can be appropriate when you
don’t need the shuffle because the processing can be carried out entirely in parallel (a
few examples are discussed in “NLineInputFormat” on page 247). In this case, the
only off-node data transfer is when the map tasks write to HDFS (see Figure 2-5).
No comments:
Post a Comment