当前位置：首页 > news >正文

Hadoop 之 Yarn (资源调度和分配)

news 2026/7/21 17:22:34

disk 不足

hadoop-hadoop-nodemanager-172.17.48.124.log

2025-07-2511:24:35,539ERROR[main]org.apache.hadoop.yarn.server.nodemanager.LocalDirsHandlerService:Most of the disks failed.1/4local-dirs have errors:[/data3/emr/yarn/local:Cannot create directory:/data3/emr/yarn/local]3/4local-dirs usable space is below configured utilization percentage/no more usable space[/data2/emr/yarn/local:used space above threshold of90.0%,/data1/emr/yarn/local:used space above threshold of90.0%,/data/emr/yarn/local:used space above threshold of90.0%];1/4log-dirs have errors:[/data3/emr/yarn/logs:Cannot create directory:/data3/emr/yarn/logs]3/4log-dirs usable space is below configured utilization percentage/no more usable space[/data/emr/yarn/logs:used space above threshold of90.0%,/data1/emr/yarn/logs:used space above threshold of90.0%,/data2/emr/yarn/logs:used space above threshold of90.0%]

maximum-container-assignments

/** Maximum number of containers to assign on each check-in. */@PrivatepublicstaticfinalStringMAX_ASSIGN_PER_HEARTBEAT=PREFIX+"per-node-heartbeat.maximum-container-assignments";

/** * Avoid potential risk that greedy assign multiple may involve * */@PrivatepublicstaticfinalintDEFAULT_MAX_ASSIGN_PER_HEARTBEAT=100;

hadoop-yarn-project\hadoop-yarn\hadoop-yarn-server\hadoop-yarn-server-resourcemanager\src\main\java\org\apache\hadoop\yarn\server\resourcemanager\scheduler\capacity\CapacitySchedulerConfiguration.java

publicintgetMaxAssignPerHeartbeat(){returngetInt(MAX_ASSIGN_PER_HEARTBEAT,DEFAULT_MAX_ASSIGN_PER_HEARTBEAT);}

hadoop-yarn-project\hadoop-yarn\hadoop-yarn-server\hadoop-yarn-server-resourcemanager\src\main\java\org\apache\hadoop\yarn\server\resourcemanager\scheduler\capacity\CapacityScheduler.java

this.maxAssignPerHeartbeat=this.conf.getMaxAssignPerHeartbeat();

// assignMultipleEnabled should be ON,// and assignedContainers should be under thresholdreturnassignMultipleEnabled&&(maxAssignPerHeartbeat==-1||assignedContainers<maxAssignPerHeartbeat);

/** * We need to make sure when doing allocation, Node should be existed * And we will construct a {@link CandidateNodeSet} before proceeding */privatevoidallocateContainersToNode(NodeIdnodeId,booleanwithNodeHeartbeat){FiCaSchedulerNodenode=getNode(nodeId);if(null!=node){intoffswitchCount=0;intassignedContainers=0;CandidateNodeSet<FiCaSchedulerNode>candidates=getCandidateNodeSet(node);CSAssignmentassignment=allocateContainersToNode(candidates,withNodeHeartbeat);// Only check if we can allocate more container on the same node when// scheduling is triggered by node heartbeatif(null!=assignment&&withNodeHeartbeat){if(assignment.getType()==NodeType.OFF_SWITCH){offswitchCount++;}if(Resources.greaterThan(calculator,getClusterResource(),assignment.getResource(),Resources.none())){assignedContainers++;}while(canAllocateMore(assignment,offswitchCount,assignedContainers)){// Try to see if it is possible to allocate multiple container for// the same node heartbeatassignment=allocateContainersToNode(candidates,true);if(null!=assignment&&assignment.getType()==NodeType.OFF_SWITCH){offswitchCount++;}if(null!=assignment&&Resources.greaterThan(calculator,getClusterResource(),assignment.getResource(),Resources.none())){assignedContainers++;}}if(offswitchCount>=offswitchPerHeartbeatLimit){LOG.debug("Assigned maximum number of off-switch containers: {},"+" assignments so far: {}",offswitchCount,assignment);}}}}

@Lock(Lock.NoLock.class)publicList<FiCaSchedulerNode>getAllNodes(){returnnodeTracker.getAllNodes();}

assignedContainers

assignedContainers 累加的代码

while(canAllocateMore(assignment,offswitchCount,assignedContainers)){// Try to see if it is possible to allocate multiple container for// the same node heartbeatassignment=allocateContainersToNode(candidates,true);if(null!=assignment&&assignment.getType()==NodeType.OFF_SWITCH){offswitchCount++;}if(null!=assignment&&Resources.greaterThan(calculator,getClusterResource(),assignment.getResource(),Resources.none())){assignedContainers++;}}

privatebooleancanAllocateMore(CSAssignmentassignment,intoffswitchCount,intassignedContainers){...// assignMultipleEnabled should be ON,// and assignedContainers should be under thresholdreturnassignMultipleEnabled&&(maxAssignPerHeartbeat==-1||assignedContainers<maxAssignPerHeartbeat);...}

nodes 来源

/** * Convenience method to filter nodes based on a condition. * * @param nodeFilter A {@link NodeFilter} for filtering the nodes * @return A list of filtered nodes */publicList<N>getNodes(NodeFilternodeFilter){List<N>nodeList=newArrayList<>();readLock.lock();try{if(nodeFilter==null){nodeList.addAll(nodes.values());}else{for(Nnode:nodes.values()){if(nodeFilter.accept(node)){nodeList.add(node);}}}}finally{readLock.unlock();}returnnodeList;}

schedule 调度随机性

/** * Schedule on all nodes by starting at a random point. * @param cs */staticvoidschedule(CapacitySchedulercs)throwsInterruptedException{// First randomize the start pointintcurrent=0;Collection<FiCaSchedulerNode>nodes=cs.nodeTracker.getAllNodes();// If nodes size is 0 (when there are no node managers registered,// we can return from here itself.intnodeSize=nodes.size();if(nodeSize==0){return;}intstart=random.nextInt(nodeSize);

CSAssignment allocateContainersToNode

CSAssignmentallocateContainersToNode(CandidateNodeSet<FiCaSchedulerNode>candidates,booleanwithNodeHeartbeat){if(rmContext.isWorkPreservingRecoveryEnabled()&&!rmContext.isSchedulerReadyForAllocatingContainers()){returnnull;}longstartTime=System.nanoTime();// Backward compatible way to make sure previous behavior which allocation// driven by node heartbeat works.FiCaSchedulerNodenode=CandidateNodeSetUtils.getSingleNode(candidates);// We have two different logics to handle allocation on single node / multi// nodes.CSAssignmentassignment;if(!multiNodePlacementEnabled){ActivitiesLogger.NODE.startNodeUpdateRecording(activitiesManager,node.getNodeID());assignment=allocateContainerOnSingleNode(candidates,node,withNodeHeartbeat);

assignment=allocateContainerOnSingleNode(candidates,node,withNodeHeartbeat);returnallocateOrReserveNewContainers(candidates,withNodeHeartbeat);

privateCSAssignmentallocateOrReserveNewContainers(CandidateNodeSet<FiCaSchedulerNode>candidates,booleanwithNodeHeartbeat){CSAssignmentassignment=getRootQueue().assignContainers(getClusterResource(),candidates,newResourceLimits(labelManager.getResourceByLabel(candidates.getPartition(),getClusterResource())),SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);assignment.setSchedulingMode(SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);submitResourceCommitRequest(getClusterResource(),assignment);