Yarn capacity scheduler
- 首先要知道, [Attention: RM有两个组件,其中Scheduler完全就只是负责资源的分配;ApplicationsManager则负责接受application,选取ApplicationMaster,监控重启AM。]
- CapacityScheduler的优点就是灵活,集群的利用率高;缺点也是由其灵活性造成的,因为CapacityScheduler不支持抢占式调度,必须等上一个任务主动释放资源。
- 可以看出,只要提交的app数没有达到上限,就能够一直提交app到集群,只是这些app会处于accepted的状态,一直pending,直到ResourceManager给它分配资源。
- 因此Scheduler实际上是会不断地schedule,然后分配资源给那些集群,在具体分配的时候,在queue内部可能会考虑优先级,然后结合FIFO。
AsyncScheduleThread
- 在CapacityScheduler内部有个AsyncScheduleThread 这个异步Scheduler线程会不断地调schedule(cs)方法。
schedule(cs)
- // Schedule on all nodes by starting at a random point.
- static void schedule(CapacityScheduler cs)
- // first randomize the start point
- Collection<FiCaSchedulerNode> nodes = cs.getAllNodes().values() // get all the nodes in the cluster
- for each node: cs.allocateContainersToNode(node)
- // assign new containers... 1. check for reserved apps(for reservation see here and here. ) 2. schedule if there are no reservations
- if reservedContainer != null
- get the reserved apps according to reservedContainers
- // try to fulfill the reservation
- LeafQueue queue = ((LeafQueue) reservedApplication.getQueue();
- CSAssignment assignment = queue.assignContainers(clusterResource, node, false); // assignContainers(Resource clusterResource, FiCaSchedulerNode node, boolean needToUnreserve);
- // if our queue cannot access this node, just return
- // check for reserved resources
- TBD...
- // try to schedule more if there are no reservations to fulfill
- if (node.getReservedContainer() == null)
- if (calculator.computeAvailableContainers(node.getAvailableResource(), minimumAllocation) > 0), then // computeAvailableContainers(Resource available, Resource required), as for DominantResourceCalculator, return the min ratio of mem & vcores.
- assignContainers(clusterResource, node, false) // assignContainers(Resource clusterResource, FiCaSchedulerNode node)
- // if our queue cannot access this node, just return
- // check for reserved resources // TBD...
- // try to assign containers to apps in order
- for (FoCaSchedulerApp application: activeApplications)
- if (SchedulerAppUtils.isBlacklisted(application, node, LOG) // check if this resource is on blacklist, i.e. cannot run the app in this node/rack
- // schedule in priority order, this is the priority of the resourceRequest of this app
- for (Priority priority : application.getPriorities())
- ResourceRequest anyRequest = application.getResourceRequest(priority, ResourceRequest.ANY)
- Resource required = anyRequest.getCapacity()
- Set<String> requestedNodeLabels = getRequestLabelSetByExpression(anyRequest.getNodeLabelExpression());
- // compute user-limit & set headroom.
- Resource userLimit = computeUserLimitAndSetHeadroom(application, clusterResource, required, requestedNodeLabels)
- // compute user limit respect requested labels
- // TODO: need consider headroom respect labels also
- Resource userLimit = computeUserLimit(application, clusterResource, required, queueUser, requestedLabels)
- // our current capacity: equal to the max(required, queue-capacity) if we're running below capacity, equal to (usedResources + required) if running over capacity.
- // if we have labels to request(choose to use the first one).
- // else if no label on request, just use absolute capacity as capacity for nodes without label.
- // TBD...
- // max avail capacity needs to take into account usage by ancestor-siblings which are greater than their base
- // calculate absoluteMaxAvailCapacity: my max avail is min(my max capacity, unused from my parent by my siblings if they are beyond their base capacity)
- // then calculate queueMaxCap using absoluteMaxAvailCapacity
- // check canAssignToThisQueue
- // consider the intersection of queue-canAccessLabels and node-labels, if any of the label beyond queue limit, we cannot allocate on this node.
- // check user limit
- application.addSchedulingOpportunity(priority);
- // try to schedule...
- TBD...
- assignContainers(clusterResource, node, false) // assignContainers(Resource clusterResource, FiCaSchedulerNode node)
- if (calculator.computeAvailableContainers(node.getAvailableResource(), minimumAllocation) > 0), then // computeAvailableContainers(Resource available, Resource required), as for DominantResourceCalculator, return the min ratio of mem & vcores.
- FYI:
/**
* Headroom is:
* min(
* min(userLimit, queueMaxCap) - userConsumed,
* queueMaxCap - queueUsedResources
* )
*
* ( which can be expressed as,
* min (userLimit - userConsumed, queuMaxCap - userConsumed,
* queueMaxCap - queueUsedResources)
* )
*
* given that queueUsedResources >= userConsumed, this simplifies to
*
* >> min (userlimit - userConsumed, queueMaxCap - queueUsedResources) <<
*
*/
addApplication
首先在CapacityScheduler随意找了个方法
synchronized addApplication(ApplicationAttemptedId applicationAttemptId, String queueName, String user)
- sanity check
- queue == null
- !queue instanceof LeafQueue
- Represents an application from the viewpoint of the scheduler. (Each running app in the RM corresponds to one instance of the FiCaScheduler class)
- FiCaSchedulerApp SchedulerApp = new FiCaSchedulerApp(applicationAttemptId, user, queue, queue.getActiveUserManager(), rmContext);
- ActiveUsersManager tracks users in the system. (An active user is defined as someone with outstanding resource requests.)
- rmContext is the context of the RM.
- submit to the queue
- try: queue.submitApplication(SchedulerApp, user, queueName)
- check queue ACLs
- synchronized(this)
- check if the queue is accepting jobs: if (getState() != QueueState.RUNNING) throw Exception
- check submission limits for queues:
- if (getNumApplications() >= getMaxApplications()) throw Exception
- check submission limits for the user on this queue
- addApplication(applictaion, user)
- user.submitApplication() : pendingApp ++; // accepted
- activateApplications():
- for each pending apps:
- check queue limit & user limit again (same as above)
- activateApplication(): --pendingApp; ++activeApp;
- for each pending apps:
- metrics.submitApp(userName, attempId): // each queue has a metrics which is an instance of QueueMetrics
- update metrics: appsSubmitted, appsFailed, appsPending
- if (parent != null) parent.submitApp(user, attemptId) // to inform the parents recursively
- try: queue.submitApplication(SchedulerApp, user, queueName)
以上,可以看到在addApplication方法内主要是判断了ACL和appNum的上限,没有resource相关的分配和判断。资源(container)的分配是由相应的applicationMaster向Resourcemanager统一请求的。ResourceRequest使用protobuf。
- 用户提交应用程序 --> ResourceManager --> ACL等检查 --> app accepted.
- 一旦Scheduler有足够的资源可以满足需求 --> app由accepted转成running --> RM为ApplicationMaster分配一个container,并负责在节点上拉起它。
- AM是每个用户作业的主进程,负责管理作业生命周期,包括动态地增加or减少资源(container),管理执行流程,处理故障和计算偏差。