[YARN-2877] Extend YARN to support distributed scheduling

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 2.9.0, 3.0.0-alpha2
• Component/s: nodemanager, resourcemanager
• Target Version/s: 2.9.0
• Hadoop Flags: Reviewed

Release Note

With this JIRA we are introducing distributed scheduling in YARN.
In particular, we make the following contributions:
- Introduce the notion of container types. GUARANTEED containers follow the semantics of the existing YARN containers. OPPORTUNISTIC ones can be seen as lower priority containers, and can be preempted in order to make space for GUARANTEED containers to run.
- Queuing of tasks at the NMs. This enables us to send more containers in an NM than its available resources. At the moment we are allowing queuing of OPPORTUNISTIC containers. Once resources become available at the NM, such containers can immediately start their execution.
- Introduce the AMRMProxy. This is a service running at each node, intercepting the requests between the AM and the RM. It is instrumental for both distributed scheduling and YARN Federation (YARN-2915).
- Enable distributed scheduling. To minimize their allocation latency, OPPORTUNISTIC containers are dispatched immediately to NMs in a distributed fashion by using the AMRMProxy of the node where the corresponding AM resides, without needing to go through the ResourceManager.

All the functionality introduced in this JIRA is disabled by default, so it will not affect the behavior of existing applications.
We have introduced parameters in YarnConfiguration to enable NM queuing (yarn.nodemanager.container-queuing-enabled), distributed scheduling (yarn.distributed-scheduling.enabled) and the AMRMProxy service (yarn.nodemanager.amrmproxy.enable).
AMs currently need to specify the type of container to be requested for each task. We are in the process of adding in the MapReduce AM the ability to randomly request OPPORTUNISTIC containers for a specified percentage of a job's tasks, so that users can experiment with the new features.

• Assignee: Konstantinos Karanasos
• Reporter: Sriram Rao
• Created: 19/Nov/14 01:17
• Updated: 25/Oct/19 20:26
• Resolved: 04/Nov/17 17:51

Description

This is an umbrella JIRA that proposes to extend YARN to support distributed scheduling. Briefly, some of the motivations for distributed scheduling are the following:
1. Improve cluster utilization by opportunistically executing tasks otherwise idle resources on individual machines.
2. Reduce allocation latency. Tasks where the scheduling time dominates (i.e., task execution time is much less compared to the time required for obtaining a container from the RM).

Issue Links

• is blocked by: YARN-2882 Add an OPPORTUNISTIC ExecutionType
• is depended upon by: MAPREDUCE-6703 Add flag to allow MapReduce AM to request for OPPORTUNISTIC containers
• is related to: YARN-5200 Improve yarn logs to get Container List
• is related to: YARN-4742 [Umbrella] Enhancements to Distributed Scheduling
• is related to: YARN-4597 Introduce ContainerScheduler and a SCHEDULED state to NodeManager container lifecycle
• is related to: YARN-5542 Scheduling of opportunistic containers
• is related to: YARN-4879 Enhance Allocate Protocol to Identify Requests Explicitly
• is related to: YARN-5646 Add documentation and update config parameter names for scheduling of OPPORTUNISTIC containers
• relates to: YARN-5688 Make allocation of opportunistic containers asynchronous
• relates to: YARN-5823 Update NMTokens in case of requests with only opportunistic containers

Sub-Tasks

• Add an OPPORTUNISTIC ExecutionType: Resolved
• Allow ResourceRequests to specify ExecutionType of a request ask: Resolved
• Proxying all AM-RM communications: Resolved
• Create AMRMProxy request interceptor for distributed scheduling decisions for queueable containers: Resolved
• Queuing of container requests in the NM: Resolved
• Extend NMStateStore to save queued container information: Resolved
• Notify the RM about the status of OPPORTUNISTIC containers: Resolved
• Create ClusterMonitor to compute ordered list of preferred NMs for OPPORTUNITIC containers: Resolved
• Corrective mechanisms for rebalancing NM container queues: Resolved
• Refactor startContainerInternal() in ContainerManager to remove unused parameter: Resolved
• Fix miscellaneous testcase errors due to YARN-2885: Resolved
• Fix findbugs warning in hadoop-yarn-common module: Resolved
• Add End-to-End test-cases for DistributedScheduling using MiniYarnCluster: Resolved
• Fix OpportunisticContainerAllocator to insert complete HostAddress in issued ContainerTokenIds: Resolved
• AM policies for choosing type of containers: Resolved