disk 不足hadoop-hadoop-nodemanager-172.17.48.124.log2025-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_HEARTBEATPREFIXper-node-heartbeat.maximum-container-assignments;/** * Avoid potential risk that greedy assign multiple may involve * */PrivatepublicstaticfinalintDEFAULT_MAX_ASSIGN_PER_HEARTBEAT100;hadoop-yarn-project\hadoop-yarn\hadoop-yarn-server\hadoop-yarn-server-resourcemanager\src\main\java\org\apache\hadoop\yarn\server\resourcemanager\scheduler\capacity\CapacitySchedulerConfiguration.javapublicintgetMaxAssignPerHeartbeat(){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.javathis.maxAssignPerHeartbeatthis.conf.getMaxAssignPerHeartbeat();// assignMultipleEnabled should be ON,// and assignedContainers should be under thresholdreturnassignMultipleEnabled(maxAssignPerHeartbeat-1||assignedContainersmaxAssignPerHeartbeat);/** * We need to make sure when doing allocation, Node should be existed * And we will construct a {link CandidateNodeSet} before proceeding */privatevoidallocateContainersToNode(NodeIdnodeId,booleanwithNodeHeartbeat){FiCaSchedulerNodenodegetNode(nodeId);if(null!node){intoffswitchCount0;intassignedContainers0;CandidateNodeSetFiCaSchedulerNodecandidatesgetCandidateNodeSet(node);CSAssignmentassignmentallocateContainersToNode(candidates,withNodeHeartbeat);// Only check if we can allocate more container on the same node when// scheduling is triggered by node heartbeatif(null!assignmentwithNodeHeartbeat){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 heartbeatassignmentallocateContainersToNode(candidates,true);if(null!assignmentassignment.getType()NodeType.OFF_SWITCH){offswitchCount;}if(null!assignmentResources.greaterThan(calculator,getClusterResource(),assignment.getResource(),Resources.none())){assignedContainers;}}if(offswitchCountoffswitchPerHeartbeatLimit){LOG.debug(Assigned maximum number of off-switch containers: {}, assignments so far: {},offswitchCount,assignment);}}}}Lock(Lock.NoLock.class)publicListFiCaSchedulerNodegetAllNodes(){returnnodeTracker.getAllNodes();}assignedContainersassignedContainers 累加的代码while(canAllocateMore(assignment,offswitchCount,assignedContainers)){// Try to see if it is possible to allocate multiple container for// the same node heartbeatassignmentallocateContainersToNode(candidates,true);if(null!assignmentassignment.getType()NodeType.OFF_SWITCH){offswitchCount;}if(null!assignmentResources.greaterThan(calculator,getClusterResource(),assignment.getResource(),Resources.none())){assignedContainers;}}privatebooleancanAllocateMore(CSAssignmentassignment,intoffswitchCount,intassignedContainers){...// assignMultipleEnabled should be ON,// and assignedContainers should be under thresholdreturnassignMultipleEnabled(maxAssignPerHeartbeat-1||assignedContainersmaxAssignPerHeartbeat);...}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 */publicListNgetNodes(NodeFilternodeFilter){ListNnodeListnewArrayList();readLock.lock();try{if(nodeFilternull){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 pointintcurrent0;CollectionFiCaSchedulerNodenodescs.nodeTracker.getAllNodes();// If nodes size is 0 (when there are no node managers registered,// we can return from here itself.intnodeSizenodes.size();if(nodeSize0){return;}intstartrandom.nextInt(nodeSize);CSAssignment allocateContainersToNodeCSAssignmentallocateContainersToNode(CandidateNodeSetFiCaSchedulerNodecandidates,booleanwithNodeHeartbeat){if(rmContext.isWorkPreservingRecoveryEnabled()!rmContext.isSchedulerReadyForAllocatingContainers()){returnnull;}longstartTimeSystem.nanoTime();// Backward compatible way to make sure previous behavior which allocation// driven by node heartbeat works.FiCaSchedulerNodenodeCandidateNodeSetUtils.getSingleNode(candidates);// We have two different logics to handle allocation on single node / multi// nodes.CSAssignmentassignment;if(!multiNodePlacementEnabled){ActivitiesLogger.NODE.startNodeUpdateRecording(activitiesManager,node.getNodeID());assignmentallocateContainerOnSingleNode(candidates,node,withNodeHeartbeat);assignmentallocateContainerOnSingleNode(candidates,node,withNodeHeartbeat);returnallocateOrReserveNewContainers(candidates,withNodeHeartbeat);privateCSAssignmentallocateOrReserveNewContainers(CandidateNodeSetFiCaSchedulerNodecandidates,booleanwithNodeHeartbeat){CSAssignmentassignmentgetRootQueue().assignContainers(getClusterResource(),candidates,newResourceLimits(labelManager.getResourceByLabel(candidates.getPartition(),getClusterResource())),SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);assignment.setSchedulingMode(SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);submitResourceCommitRequest(getClusterResource(),assignment);代码调用顺序是schedule cs.allocateContainersToNodeAI 辅助分析