Tom White - Hadoop The Definitive Guide_ 4 edition - 2015 (811394), страница 42
Текст из файла (страница 42)
A task attempt maybe killed because it is a speculative duplicate (for more information on this topic, see“Speculative Execution” on page 204), or because the node manager it was running onfailed and the application master marked all the task attempts running on it as killed.Killed task attempts do not count against the number of attempts to run the task (as setby mapreduce.map.maxattempts and mapreduce.reduce.maxattempts), because itwasn’t the task’s fault that an attempt was killed.Users may also kill or fail task attempts using the web UI or the command line (typemapred job to see the options).
Jobs may be killed by the same mechanisms.Application Master FailureJust like MapReduce tasks are given several attempts to succeed (in the face of hardwareor network failures), applications in YARN are retried in the event of failure. The max‐imum number of attempts to run a MapReduce application master is controlled by themapreduce.am.max-attempts property. The default value is 2, so if a MapReduce ap‐plication master fails twice it will not be tried again and the job will fail.YARN imposes a limit for the maximum number of attempts for any YARN applicationmaster running on the cluster, and individual applications may not exceed this limit.The limit is set by yarn.resourcemanager.am.max-attempts and defaults to 2, so ifyou want to increase the number of MapReduce application master attempts, you willhave to increase the YARN setting on the cluster, too.The way recovery works is as follows. An application master sends periodic heartbeatsto the resource manager, and in the event of application master failure, the resourcemanager will detect the failure and start a new instance of the master running in a newcontainer (managed by a node manager).
In the case of the MapReduce application194| Chapter 7: How MapReduce Worksmaster, it will use the job history to recover the state of the tasks that were already runby the (failed) application so they don’t have to be rerun. Recovery is enabled by default,but can be disabled by setting yarn.app.mapreduce.am.job.recovery.enable tofalse.The MapReduce client polls the application master for progress reports, but if its ap‐plication master fails, the client needs to locate the new instance.
During job initializa‐tion, the client asks the resource manager for the application master’s address, and thencaches it so it doesn’t overload the resource manager with a request every time it needsto poll the application master. If the application master fails, however, the client willexperience a timeout when it issues a status update, at which point the client will goback to the resource manager to ask for the new application master’s address. Thisprocess is transparent to the user.Node Manager FailureIf a node manager fails by crashing or running very slowly, it will stop sending heartbeatsto the resource manager (or send them very infrequently).
The resource manager willnotice a node manager that has stopped sending heartbeats if it hasn’t received one for10minutes(thisisconfigured,inmilliseconds,viatheyarn.resourcemanager.nm.liveness-monitor.expiry-interval-ms property) andremove it from its pool of nodes to schedule containers on.Any task or application master running on the failed node manager will be recoveredusing the mechanisms described in the previous two sections. In addition, the applica‐tion master arranges for map tasks that were run and completed successfully on thefailed node manager to be rerun if they belong to incomplete jobs, since their inter‐mediate output residing on the failed node manager’s local filesystem may not be ac‐cessible to the reduce task.Node managers may be blacklisted if the number of failures for the application is high,even if the node manager itself has not failed. Blacklisting is done by the applicationmaster, and for MapReduce the application master will try to reschedule tasks on dif‐ferent nodes if more than three tasks fail on a node manager.
The user may set thethreshold with the mapreduce.job.maxtaskfailures.per.tracker job property.Note that the resource manager does not do blacklisting across ap‐plications (at the time of writing), so tasks from new jobs may bescheduled on bad nodes even if they have been blacklisted by anapplication master running an earlier job.Failures|195Resource Manager FailureFailure of the resource manager is serious, because without it, neither jobs nor taskcontainers can be launched. In the default configuration, the resource manager is asingle point of failure, since in the (unlikely) event of machine failure, all running jobsfail—and can’t be recovered.To achieve high availability (HA), it is necessary to run a pair of resource managers inan active-standby configuration.
If the active resource manager fails, then the standbycan take over without a significant interruption to the client.Information about all the running applications is stored in a highly available state store(backed by ZooKeeper or HDFS), so that the standby can recover the core state of thefailed active resource manager. Node manager information is not stored in the statestore since it can be reconstructed relatively quickly by the new resource manager asthe node managers send their first heartbeats.
(Note also that tasks are not part of theresource manager’s state, since they are managed by the application master. Thus, theamount of state to be stored is therefore much more manageable than that of the job‐tracker in MapReduce 1.)When the new resource manager starts, it reads the application information from thestate store, then restarts the application masters for all the applications running on thecluster. This does not count as a failed application attempt (so it does not count againstyarn.resourcemanager.am.max-attempts), since the application did not fail due to anerror in the application code, but was forcibly killed by the system.
In practice, theapplication master restart is not an issue for MapReduce applications since they recoverthe work done by completed tasks (as we saw in “Application Master Failure” on page194).The transition of a resource manager from standby to active is handled by a failovercontroller. The default failover controller is an automatic one, which uses ZooKeeperleader election to ensure that there is only a single active resource manager at one time.Unlike in HDFS HA (see “HDFS High Availability” on page 48), the failover controllerdoes not have to be a standalone process, and is embedded in the resource manager bydefault for ease of configuration. It is also possible to configure manual failover, but thisis not recommended.Clients and node managers must be configured to handle resource manager failover,since there are now two possible resource managers to communicate with.
They tryconnecting to each resource manager in a round-robin fashion until they find the activeone. If the active fails, then they will retry until the standby becomes active.196|Chapter 7: How MapReduce WorksShuffle and SortMapReduce makes the guarantee that the input to every reducer is sorted by key. Theprocess by which the system performs the sort—and transfers the map outputs to thereducers as inputs—is known as the shuffle.4 In this section, we look at how the shuffleworks, as a basic understanding will be helpful should you need to optimize a MapRe‐duce program.
The shuffle is an area of the codebase where refinements andimprovements are continually being made, so the following description necessarilyconceals many details. In many ways, the shuffle is the heart of MapReduce and is wherethe “magic” happens.The Map SideWhen the map function starts producing output, it is not simply written to disk. Theprocess is more involved, and takes advantage of buffering writes in memory and doingsome presorting for efficiency reasons.
Figure 7-4 shows what happens.Figure 7-4. Shuffle and sort in MapReduceEach map task has a circular memory buffer that it writes the output to. The buffer is100 MB by default (the size can be tuned by changing the mapreduce.task.io.sort.mbproperty). When the contents of the buffer reach a certain threshold size (mapreduce.map.sort.spill.percent, which has the default value 0.80, or 80%), a back‐ground thread will start to spill the contents to disk. Map outputs will continue to bewritten to the buffer while the spill takes place, but if the buffer fills up during this time,4.
The term shuffle is actually imprecise, since in some contexts it refers to only the part of the process wheremap outputs are fetched by reduce tasks. In this section, we take it to mean the whole process, from the pointwhere a map produces output to where a reduce consumes input.Shuffle and Sort|197the map will block until the spill is complete. Spills are written in round-robin fashionto the directories specified by the mapreduce.cluster.local.dir property, in a jobspecific subdirectory.Before it writes to disk, the thread first divides the data into partitions correspondingto the reducers that they will ultimately be sent to. Within each partition, the backgroundthread performs an in-memory sort by key, and if there is a combiner function, it is runon the output of the sort.
![](https://s.studizba.com/z.php?f=/templates/si/i/noavatar.png&w=100&h=100&t=1"){kind=avatar}
