diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java index a54104d..883058a 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java @@ -377,7 +377,8 @@ public static boolean isAclEnabled(Configuration conf) { public static final long NM_CONTAINER_QUEUING_SORTING_NODES_INTERVAL_MS_DEFAULT = 1000; - /** Comparator for determining node load for Distributed Scheduling. */ + /** Comparator for determining node load for scheduling of opportunistic + * containers. */ public static final String NM_CONTAINER_QUEUING_LOAD_COMPARATOR = YARN_PREFIX + "nm-container-queuing.load-comparator"; public static final String NM_CONTAINER_QUEUING_LOAD_COMPARATOR_DEFAULT = diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AMRMProxyService.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AMRMProxyService.java index 79882aa..dc56090 100644 --- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AMRMProxyService.java +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AMRMProxyService.java @@ -79,7 +79,7 @@ * to intercept and inspect messages from application master to the cluster * resource manager. It listens to messages from the application master and * creates a request intercepting pipeline instance for each application. The - * pipeline is a chain of intercepter instances that can inspect and modify the + * pipeline is a chain of interceptor instances that can inspect and modify the * request/response as needed. */ public class AMRMProxyService extends AbstractService implements diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/OpportunisticContainers.md hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/OpportunisticContainers.md new file mode 100644 index 0000000..3ae0b19 --- /dev/null +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/OpportunisticContainers.md @@ -0,0 +1,235 @@ + + +Opportunistic Containers +======================== + +* [Purpose](#Purpose) +* [Overview](#Overview) +* [Container Execution Types](#Container_Execution_Types) +* [Allocation of Opportunistic Containers](#Opportunistic_Allocation) + * [Centralized Allocation](#Centralized_Allocation) + * [Distributed Allocation](#Distributed_Allocation) + * [Determining Nodes for Allocation](#Nodes_to_Allocate) + * [Rebalancing Node Load](#Node_Load_Rebalancing) +* [Execution of Opportunistic Containers](#Opportunistic_Execution) +* [Configuration](#Configuration) + * [Basic Configuration](#Basic_Configuration) + * [Configuring Execution Properties of Opportunistic Containers](#Configure_Opportunistic_Execution) + * [Configuring Allocation Properties of Opportunistic Containers](#Configure_Opportunistic_Allocation) +* [Running a Sample Job with Opportunistic Containers](#Sample_Job_Opportunistic_Containers) +* [Opportunistic Containers in the Web UI](#Web_UI) +* [Open Items for Future Work](#Future_Work) + + +Purpose +----------------------------- + +This document introduces the notion of **opportunistic** container execution, and discusses how opportunistic containers are allocated and executed. + + +Overview +-------------------------------- + +The existing schedulers in YARN (Fair and Capacity Scheduler) allocate containers to a node only if there are available resources at that node the moment of scheduling the containers. This **guaranteed** type of execution has the advantage that once the AM dispatches a container to a node, the container execution will start immediately, since it is guaranteed that there will be available resources. Moreover, unless fairness or capacity constraints are violated, containers are guaranteed to run to completion without being preempted. + +Although this design offers a more predictable task execution, it has two main drawbacks that can lead to suboptimal cluster resource utilization: + +* **Feedback delays.** When a container finishes its execution at a node, the RM gets notified that there are available resources through the next NM-RM heartbeat, then the RM schedules a new container at that node, the AM gets notified through the next AM-RM heartbeat, and finally the AM launches the new container at the node. These delays result in idle node resources, which in turn lead to lower resource utilization, especially for tasks whose duration is relatively short. +* **Allocated vs. utilized resources.** The RM allocates containers based on the *allocated* resources at each node, which might be significantly higher than the actually *utilized* resources (e.g., think of a container for which 4GB memory have been allocated, but only 2GB are being utilized). This lowers effective resource utilization, and can be avoided if the RM takes into account the utilized resources during scheduling. However, this has to be done in a way that allows resources to be reclaimed in case the utilized resources of a running container increase. + +To mitigate the above problems, we introduce the **opportunistic** container execution type, along with the already existing guaranteed one. An opportunistic container can be dispatched to an NM, even if there are no available (unallocated) resources for it at the moment of scheduling. In such a case, the opportunistic container will be queued at the NM, waiting for resources to become available for its execution to start. The opportunistic containers are of lower priority than the guaranteed ones, which means that they can be preempted/killed for guaranteed containers to start their execution. Therefore, they can be used to improve cluster resource utilization without impacting the execution of existing guaranteed containers. + +An additional advantage of opportunistic containers is that they introduce a notion of **execution priority at the NMs**. For instance, a lower priority job that does not require strict execution guarantees can use opportunistic containers or a mix of container execution types for its tasks. + +We have introduced two ways of allocating opportunistic containers: a **centralized** and a **distributed** one. In the centralized scheduling, opportunistic containers are allocated through the YARN RM, whereas in the distributed one, through local schedulers that reside at each NM. Centralized allocation allows for higher quality placement decisions and for implementing more involved sharing policies across applications (e.g., fairness). On the other hand, distributed scheduling can offer faster container allocation, which is useful for short tasks, as it avoids the round-trip to the RM. In both cases, the scheduling of guaranteed containers remains intact and happens through the YARN RM (using the existing Fair or Capacity Scheduler). + +Note that in the current implementation, we are allocating containers based on allocated (and not utilized) resources. Therefore, we tackle the "feedback delays" problem mentioned above, but not the "allocated vs. utilized resources" one. There is ongoing work (`YARN-1011`) that employs opportunistic containers to address the latter problem too. + +Below, we describe in more detail the [container execution types](#Container_Execution_Types), as well as the [allocation](#Opportunistic_Allocation) and [execution](#Opportunistic_Execution) (including the container queuing at the NMs) of opportunistic containers. Then we discuss the required [configuration](#Configure_Opportunistic_Containers) for enabling opportunistic containers, how to run some [sample jobs](#Sample_Job_Opportunistic_Containers), and how to observe opportunistic containers in the [Web UI](#Web_UI). Finally, we discuss open items for [future work](#Future_Work). + + +Container Execution Types +----------------------------------------------------------------- + +We introduce the following two types of containers: + +* **Guaranteed containers** correspond to the existing YARN containers. They are allocated by the Fair or Capacity Scheduler, and once dispatched to a node, it is guaranteed that there are available resources for their execution to start immediately. Moreover, these containers run to completion (as long as there are no failures). They can be preempted only in case the scheduler's queue to which they belong, violates fairness or capacity constraints. +* **Opportunistic containers** are not guaranteed to have resources for their execution to start when they get dispatched to a node. Instead, they might be queued at the NM until resources become available. In case a guaranteed container arrives at a node and there are no resources available for it, a number of opportunistic containers will be pre-empted/killed to execute the guaranteed one. + +When an AM submits its resource requests to the RM, it specifies the type for each container (default is guaranteed), determining the way the container will be [allocated](#Opportunistic_Allocation). Subsequently, when the container is launched by the AM at an NM, its type determines how it will be [executed](#Opportunistic_Execution) by the NM. + + + +Allocation of Opportunistic Containers +----------------------------------------------------------------------------- + +As mentioned above, we provide both a centralized and a distributed way of allocating opportunistic containers, which we describe below. + +###Centralized Allocation + +We have introduced a new service at the RM, namely the `OpportunisticContainerAllocatorAMService`, which extends the `ApplicationMasterService`. When the centralized opportunistic allocation is enabled, the resource requests from the AMs are served at the RM side by the `OpportunisticContainerAllocatorAMService`, which splits them into two sets of resource requests: + +* The guaranteed set is forwarded to the existing `ApplicationMasterService` and is subsequently handled by the Fair or Capacity Scheduler. +* The opportunistic set is handled by the new `OpportunisticContainerAllocator`, which performs the scheduling of opportunistic containers to nodes. + +The `OpportunisticContainerAllocator` maintains a list with the [least loaded nodes](#Nodes_to_Allocate) of the cluster at each moment, and assigns containers to them in a round-robin fashion. Note that in the current implementation, we purposely do not take into account node locality constraints. Since an opportunistic container (unlike the guaranteed ones) might wait at the queue of an NM before its execution starts, it is more important to allocate it at a node that is less loaded (i.e., where queuing delay will be smaller) rather than respect its locality constraints. Moreover, we do not take into account sharing (fairness/capacity) constraints for opportunistic containers at the moment. Support for both locality and sharing constraints can be added in the future if required. + + +###Distributed Allocation + +In order to enable distributed scheduling of opportunistic containers, we have introduced a new service at each NM, called `AMRMProxyService`. The `AMRMProxyService` implements the `ApplicationMasterService` protocol, and acts as a proxy between the AMs running at that node and the RM. When the `AMRMProxyService` is enabled (through a parameter), we force all AMs running at a particular node to communicate with the `AMRMProxyService` of the same node, instead of going directly to the RM. Moreover, to ensure that the AMs will not talk directly with the RM, when a new AM gets initialized, we replace its `AMRMToken` with a token signed by the `AMRMProxyService`. + +A chain of interceptors can be registered with the `AMRMProxyService`. One of these interceptors is the `DistributedScheduler` that is responsible for allocating opportunistic containers in a distributed way, without needing to contact the RM. This modular design makes the `AMRMProxyService` instrumental in other scenarios too, such as YARN federation (`YARN-2915`) or throttling down misbehaving AMs, which can be enabled simply by adding additional interceptors at the interceptor chain. + +When distributed opportunistic scheduling is enabled, each AM sends its resource requests to the `AMRMProxyService` running at the same node. The `AMRMProxyService` splits the resource requests into two sets: + +* The guaranteed set is forwarded to the RM. In this case the `AMRMProxyService` simply acts as a proxy between the AM and the RM, and the container allocation remains intact (using the Fair or Capacity Scheduler). +* The opportunistic set is not forwarded to the RM. Instead, it is handled by the `DistributedScheduler` that is running locally at the node. In particular, the `DistributedScheduler` maintains a list with the list loaded nodes in the cluster, and allocates containers to them in a round-robin fashion. The RM informs the `DistributedScheduler` about the least loaded nodes at regular intervals through the NM-RM heartbeats. + +The above procedure is similar to the one performed by the `OpportunisticContainerAllocatorAMService` in the case of centralized opportunistic scheduling described above. The main difference is that in the distributed case, the splitting of requests into guaranteed and opportunistic happens locally at the node, and only the guaranteed requests are forwarded to the RM, while the opportunistic ones are handled without contacting the RM. + + +###Determining Nodes for Allocation + +Each NM informs the RM periodically through the NM-RM heartbeats about the number of running guaranteed and opportunistic containers, as well as the number of queued opportunistic containers. The RM gathers this information from all nodes and determines the least loaded ones. + +In case of centralized allocation of opportunistic containers, this information is immediately available, since the allocation happens centrally. In case of distributed scheduling, the list with the least loaded nodes is propagated to all NMs (and thus becomes available to the `DistributedSchedulers`) through the heartbeat responses from the RM to the NMs. The number of least loaded nodes sent to the NMs is configurable. + +At the moment, we take into account only the number of queued opportunistic containers at each node in order to estimate the time an opportunistic container would have to wait if sent to that node and, thus, determine the least loaded nodes. If the AM provided us with information about the estimated task durations, we could take them into account in order to have better estimates of the queue waiting times. + + +###Rebalancing Node Load + +Occasionally poor placement choices for opportunistic containers may be made (due to imprecise or stale queue time estimates), which can lead to load imbalance between the nodes. The problem is more pronounced under high cluster load, and also in case of distributed scheduling (multiple `DistributedSchedulers` may place containers at the same NM, since they do not coordinate with each other). To deal with this load imbalance between the NM queues, we perform load shedding to dynamically re-balance the load between the NMs. In particular, while aggregating at the RM the queue time estimates published by each NM, we construct a distribution and find a targeted maximal value for the length of the NM queues (based on the mean and standard deviation of the distribution). Then the RM disseminates this value to the various NMs through the heartbeat responses. Subsequently, using this information, the NM on a node whose queue length is above the threshold, discards opportunistic containers to meet this maximal value. This forces the associated individual AMs to requeue those containers elsewhere. + + +Execution of Opportunistic Containers +--------------------------------------------------------------------------- + +When a container arrives at an NM, its execution is determined by the available resources at the NM and the container type. Guaranteed containers start their execution immediately, and if needed, the NM will kill running opportunistic containers to ensure there are sufficient resources for the guaranteed ones to start. On the other hand, opportunistic containers can be queued at the NM, if there are no resources available to start their execution when they arrive at the NM. To enable this, we extended the NM by allowing queuing of containers at each node. The NM monitors the local resources, and when there are sufficient resources available, it starts the execution of the opportunistic container that is at the head of the queue. + +In particular, when a container arrives at an NM, first localization is performed (i.e., all required resources are downloaded), and then the container moves to a `SCHEDULED` state, in which the container is queued, waiting for its execution to begin: + +* If there are available resources, the execution of the container starts immediately, irrespective of its execution type. +* If there are no available resources: + * If the container is guaranteed, we kill as many running opportunistic containers as required for the guaranteed container to be executed, and then start its execution. + * If the container is opportunistic, it remains at the queue until resources become available. +* When a container (guaranteed or opportunistic) finishes its execution and resources get freed up, we examine the queued containers and if there are available resources we start their execution. We pick containers from the queue in a FIFO order. + +In the [future work items](#Future_Work) below, we discuss different ways of prioritizing task execution (queue reordering) and of killing opportunistic containers to make space for guaranteed ones. + + + +Configuration +----------------------------------------- + +We now present all the properties that are related to opportunistic containers, and can be set at **conf/yarn-site.xml**. + +###Basic Configuration + +To enable the allocation of opportunistic containers, the following parameters have to be set to `true`: + +| Property | Value | +|:-------- |:----- | +| `yarn.opportunistic-container-allocation.enabled` | Enables opportunistic container allocation. | +| `yarn.nodemanager.opportunistic-containers-max-queue-length` | Determines the max length of the NM queue for opportunistic containers. A positive value is required to enable queuing, and each NM can set its own length. | + +Setting only the above parameters will result in [centralized allocation](#Centralized_Allocation) of opportunistic containers. To enable [distributed allocation](#Distributed_Allocation), the following two parameters have to be set too (along with the previous two): + +| Property | Value | +|:-------- |:----- | +| `yarn.nodemanager.amrmproxy.enabled` | Enables AMRMProxy at each NM. | +| `yarn.distributed-scheduling.enabled` | Enables distributed scheduling. | + +The properties we describe in the rest of the section can be used to further tune the opportunistic container allocation and execution (although the default values should be sufficient for someone to start using opportunistic containers). + + +###Configuring Opportunistic Container Properties + +The following parameters configure some of the opportunistic container properties during allocation. In particular, all but the last one are used to constrain the resource demands of an opportunistic container, that is, not allow it to use too few or too many resources: + +| Property | Value | +|:-------- |:----- | +| `yarn.opportunistic-containers.min-memory-mb` | Minimum memory (in MB) used for allocating an opportunistic container. | +| `yarn.opportunistic-containers.min-vcores` | Minimum virtual CPU cores used for allocating an opportunistic container. | +| `yarn.opportunistic-containers.max-memory-mb` | Maximum memory (in MB) used for allocating an opportunistic container. | +| `yarn.opportunistic-containers.max-vcores` | Maximum virtual CPU cores used for allocating an opportunistic container. | +| `yarn.opportunistic-containers.incr-memory-mb` | Incremental memory (in MB) used for allocating an opportunistic container. | +| `yarn.opportunistic-containers.incr-vcores` | Incremental virtual CPU cores used for allocating an opportunistic container. | +| `yarn.opportunistic-containers.container-token-expiry-ms` | Container token expiry for opportunistic containers. | + +The following properties are related to how we choose the [nodes for allocating](#Nodes_to_Allocate) opportunistic containers: + +| Property | Value | +|:-------- |:----- | +| `yarn.opportunistic-container-allocation.nodes-used` | Number of nodes to be used by the Opportunistic Container allocator for dispatching containers during container allocation. | +| `yarn.nm-container-queuing.sorting-nodes-interval-ms` | Frequency for computing least loaded nodes. | +| `yarn.nm-container-queuing.load-comparator` | Comparator for determining the load of each node and in turn choosing the least loaded nodes for scheduling of opportunistic containers. `QUEUE_LENGTH` and `QUEUE_WAIT_TIME` are currently the allowed values. The former is using the length of the NM queue to determine a node's load, whereas the latter uses the estimated queue wait time. At the moment, we support only the `QUEUE_LENGTH` load. | + | `yarn.nm-container-queuing.queue-limit-stdev` | Value of standard deviation used for calculation of queue limit thresholds. This is required for the [node load balancing](#Node_Load_Rebalancing). | + +Finally, some additional properties for the `AMRMProxyService` in case distributed scheduling is used: + +| Property | Value | +|:-------- |:----- | +| `yarn.nodemanager.amrmproxy.address` | The address/port to which the `AMRMProxyService` is bound to. | +| `yarn.nodemanager.amrmproxy.client.thread-count` | The number of threads that are used at each NM for serving the interceptors register to the `AMRMProxyService` by different jobs. | + + +###Configuring Execution Properties of Opportunistic Containers + +The properties below are related to the execution of opportunistic containers at the NMs: + +| Property | Value | +|:-------- |:----- | +| `yarn.nm-container-queuing.min-queue-length` | Min length of container queue at NM. This is the cluster-wide min queue length that all NMs have to respect. | +| `yarn.nm-container-queuing.max-queue-length` | Max length of container queue at NM. This is the cluster-wide max queue length that all NMs have to respect. Note that each NM can set its own stricter limit through `yarn.nodemanager.opportunistic-containers-max-queue-length`. | +| `yarn.nm-container-queuing.min-queue-wait-time-ms` | Min queue wait time for a container at NM. It is used when the value of `yarn.nm-container-queuing.load-comparator` is `QUEUE_WAIT_TIME`. | +| `yarn.nm-container-queuing.max-queue-wait-time-ms` | Max queue wait time for a container queue at NM. Similar to `yarn.nm-container-queuing.min-queue-wait-time-ms`. | + + +Running a Sample Job with Opportunistic Containers +---------------------------------------------------------------------------------------------------- + +In order to be able to run map-reduce jobs with opportunistic containers, we extended the `RMContainerAllocator` by adding parameter `mapreduce.job.num-opportunistic-maps-per-100`. This parameter takes an integer value that determines which percentage of the mappers of a map-reduce job will be executed as opportunistic containers. For example, a value of 40 means that 40 out of 100 mappers will use opportunistic containers. To avoid having reducers being killed, we use only guaranteed containers for reducers. Moreover, if a map task fail, for the second attempt we will use a guaranteed container. + +The following command can be used to run a sample pi map-reduce job, executing 40% of the mappers as opportunistic containers (substitute `3.0.0-alpha2-SNAPSHOT` below with the version of Hadoop you are using): +``` +$ hadoop jar hadoop-3.0.0-alpha2-SNAPSHOT/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.0.0-alpha2-SNAPSHOT.jar pi -Dmapreduce.job.num-opportunistic-maps-per-100="40" 50 100 +``` + +Opportunistic Containers in the Web UI +----------------------------------------------------------- + +When opportunistic container allocation is enabled, the following new columns can be observed in the Nodes page of the Web UI (`rm-address:8088/cluster/nodes`): +* Running Containers (O): number of running opportunistic containers on each node; +* Mem Used (O): Total memory used by opportunistic containers on each node; +* VCores Used (O): Total CPU virtual cores used by opportunistic containers on each node; +* Queued Containers: Number of containers queued at each node. + +Also, when clicking on a specific container running on a node, the execution type of the container is also shown. + + +Open Items for Future Work +----------------------------------------------- + +Here we describe multiple ways in which we can extend/enhance the allocation and execution of opportunistic containers. We also provide the open JIRAs that track each item. + +* **Resource overcommitment** (`YARN-1011`). As already discussed, in order to further improve the cluster resource utilization, we can schedule containers not based on the allocated but on the actually utilized resources. When over-committing resources, there is the risk of running out of resources in case we have an increase in the utilized resources of the already running containers. Therefore, opportunistic execution should be used for containers whose allocation goes beyond the capacity of a node. This way, we can choose opportunistic containers to kill for reclaiming resources. +* **NM Queue reordering** (`YARN-5886`). Instead of executing queued containers in a FIFO order, we can employ reordering strategies that dynamically determine which opportunistic container will be executed next. For example, we can prioritize containers that are expected to be short-running or which belong to applications that are close to completion. +* **Out of order killing at NMs** (`YARN-5887`). As described above, when we need to free up resources for a guaranteed container to start its execution, we kill opportunistic containers in reverse order of arrival (first the most recently started ones). This might not always be the right decision. For example, we might want to minimize the number of containers killed or to refrain from killing containers of jobs that are very close to completion. +* **Container preemption** (`YARN-5292`): At the moment we kill opportunistic containers to make room for guaranteed in case of resource contention. In busy clusters this can lower the effective cluster utilization, since we lose work whenever we kill a running opportunistic container, which has to be restarted. To this end, we can instead preempt running opportunistic containers. Note that this will require support from the container executor (e.g., the container technology used) and from the application. +* **Container promotion** (`YARN-5085`). There are cases where changing the execution type of a container during its execution can be beneficial. For instance, an application might submit a container as opportunistic, and when its execution starts, it can request its promotion to a guaranteed container to avoid it getting killed. + +