标签：des blog http os io java ar for 2014
AppMaster向RM请求资源
//APPMASTER向RM发送心跳，更新资源请求结构，从分配好的内存结构取出分配的资源,具体任务分配是后台异步的由NM发送心跳驱动
MRAppMaster
:serviceinit
// service to allocate containers from RM (if non-uber) or to fake it (uber)
containerAllocator = createContainerAllocator(null, context);
addIfService(containerAllocator);
dispatcher.register(ContainerAllocator.EventType.class, containerAllocator);
protected ContainerAllocator createContainerAllocator(
final ClientService clientService, final AppContext context) {
return new ContainerAllocatorRouter(clientService, context); //
}
private final class ContainerAllocatorRouter extends AbstractService
implements ContainerAllocator, RMHeartbeatHandler {
private final ClientService clientService;
private final AppContext context;
private ContainerAllocator containerAllocator;
.....
@Override
protected void serviceStart() throws Exception {
if (job.isUber()) {
this.containerAllocator = new LocalContainerAllocator(
this.clientService, this.context, nmHost, nmPort, nmHttpPort
, containerID);
} else {
this.containerAllocator = new RMContainerAllocator( ///
this.clientService, this.context);
}
((Service)this.containerAllocator).init(getConfig());
((Service)this.containerAllocator).start();
super.serviceStart();
org.apache.hadoop.mapreduce.v2.app.rm; RMContainerAllocator类有该方法
protected synchronized void heartbeat() throws Exception {
scheduleStats.updateAndLogIfChanged("Before Scheduling: ");
List<Container> allocatedContainers = getResources(); //发远程RM发送心跳信息,注意心跳里可能没有新的资源请求信息
//只是告诉RM自己还活着,或者只是从RM取得分配资源
if (allocatedContainers.size() > 0) {
scheduledRequests.assign(allocatedContainers); //获得的container具体分配到任务task (应该是重排序）
}
资源请求包括的字段:
优先级,期望在的host,内存大小等 (默认三路复制,可能会有7个资源请求,3个local,3个 rack,1个随机)
}
RMContainerAllocator父类RMCommunicator的方法
protected void startAllocatorThread() {
allocatorThread = new Thread(new Runnable() {
@Override
public void run() {
while (!stopped.get() && !Thread.currentThread().isInterrupted()) {
try {
Thread.sleep(rmPollInterval); //默认每秒
try {
heartbeat(); //发送心跳
...
private List<Container> getResources() throws Exception {
int headRoom = getAvailableResources() != null
? getAvailableResources().getMemory() : 0;//first time it would be null
AllocateResponse response;
/*
* If contact with RM is lost, the AM will wait MR_AM_TO_RM_WAIT_INTERVAL_MS
* milliseconds before aborting. During this interval, AM will still try
* to contact the RM.
*/
try {
response = makeRemoteRequest(); //关键
makeRemoteRequest方法为其父类RMContainerRequestor定义的方法
protected AllocateResponse makeRemoteRequest() throws IOException {
ResourceBlacklistRequest blacklistRequest =
ResourceBlacklistRequest.newInstance(new ArrayList<String>(blacklistAdditions),
new ArrayList<String>(blacklistRemovals));
AllocateRequest allocateRequest = //新建个资源请求
AllocateRequest.newInstance(lastResponseID,
super.getApplicationProgress(), new ArrayList<ResourceRequest>(ask), //这个ask是集合类,存ResourceRequest实例,
//只有个新建方法,在哪赋值的呢
new ArrayList<ContainerId>(release), blacklistRequest);
AllocateResponse allocateResponse;
try {
allocateResponse = scheduler.allocate(allocateRequest); //关键,分配资源,此处的scheduler 并非是调度器
//而是ApplicationMasterProtocol,他会终调用到调度器
scheduler为其父类RMCommunicator新建
protected ApplicationMasterProtocol scheduler;
...
protected void serviceStart() throws Exception {
scheduler= createSchedulerProxy();
..
protected ApplicationMasterProtocol createSchedulerProxy() {
final Configuration conf = getConfig();
try {
return ClientRMProxy.createRMProxy(conf, ApplicationMasterProtocol.class); //ApplicationMasterProtocol协议是关键
//通过他远程调用ApplicationMasterService中的方法
} catch (IOException e) {
throw new YarnRuntimeException(e);
}
}
//后面追踪ask的的赋值最终是在哪里调用
//ask的赋值方法，最后是由 addContainerReq方法，该方法在RMContainerAllocator调用
private void addResourceRequestToAsk(ResourceRequest remoteRequest) {
// because objects inside the resource map can be deleted ask can end up
// containing an object that matches new resource object but with different
// numContainers. So exisintg values must be replaced explicitly
if(ask.contains(remoteRequest)) {
ask.remove(remoteRequest);
}
ask.add(remoteRequest);
}
protected void addContainerReq(ContainerRequest req) {
// Create resource requests
for (String host : req.hosts) {
// Data-local
if (!isNodeBlacklisted(host)) {
addResourceRequest(req.priority, host, req.capability);
}
}
// Nothing Rack-local for now
for (String rack : req.racks) {
addResourceRequest(req.priority, rack, req.capability);
}
// Off-switch
addResourceRequest(req.priority, ResourceRequest.ANY, req.capability);
}
RMContainerAllocator内
void addMap(ContainerRequestEvent event) { //addMap方法
ContainerRequest request = null;
if (event.getEarlierAttemptFailed()) {
earlierFailedMaps.add(event.getAttemptID());
request = new ContainerRequest(event, PRIORITY_FAST_FAIL_MAP);
LOG.info("Added "+event.getAttemptID()+" to list of failed maps");
} else {
for (String host : event.getHosts()) {
LinkedList<TaskAttemptId> list = mapsHostMapping.get(host);
if (list == null) {
list = new LinkedList<TaskAttemptId>();
mapsHostMapping.put(host, list);
}
list.add(event.getAttemptID());
if (LOG.isDebugEnabled()) {
LOG.debug("Added attempt req to host " + host);
}
}
for (String rack: event.getRacks()) {
LinkedList<TaskAttemptId> list = mapsRackMapping.get(rack);
if (list == null) {
list = new LinkedList<TaskAttemptId>();
mapsRackMapping.put(rack, list);
}
list.add(event.getAttemptID());
if (LOG.isDebugEnabled()) {
LOG.debug("Added attempt req to rack " + rack);
}
}
request = new ContainerRequest(event, PRIORITY_MAP);
}
maps.put(event.getAttemptID(), request);
addContainerReq(request); //调用
//addMap在该方法内被调用
protected synchronized void handleEvent(ContainerAllocatorEvent event) {
recalculateReduceSchedule = true;
..................
scheduledRequests.addMap(reqEvent);//maps are immediately scheduled
protected void serviceStart() throws Exception {
this.eventHandlingThread = new Thread() {
@SuppressWarnings("unchecked")
@Override
public void run() {
ContainerAllocatorEvent event;
while (!stopped.get() && !Thread.currentThread().isInterrupted()) {
try {
event = RMContainerAllocator.this.eventQueue.take(); //取出事件
} catch (InterruptedException e) {
if (!stopped.get()) {
LOG.error("Returning, interrupted : " + e);
}
return;
}
try {
handleEvent(event); //调用
// 事件加入在MRAppMaster内,加入的事件在上面的方法被处理，该方法在哪里调用了呢？
public void handle(ContainerAllocatorEvent event) {
this.containerAllocator.handle(event);
}
//总结,applicationmaster最终通过ApplicationMasterProtocol#allocate向RM汇报资源需求,RM端的ApplicationMasterService提供服务,并最终调用调度器的allocate
//将新的资源需求写入内存结构,并返回已经分配的资源
public class ApplicationMasterService extends AbstractService implements
ApplicationMasterProtocol {
public AllocateResponse allocate(AllocateRequest request)
throws YarnException, IOException {
..
// Allow only one thread in AM to do heartbeat at a time.
synchronized (lastResponse) {
// Send the status update to the appAttempt.
this.rmContext.getDispatcher().getEventHandler().handle(
new RMAppAttemptStatusupdateEvent(appAttemptId, request
.getProgress()));
List<ResourceRequest> ask = request.getAskList(); //ask,release为封装的请求
List<ContainerId> release = request.getReleaseList(
// Send new requests to appAttempt.
Allocation allocation =
this.rScheduler.allocate(appAttemptId, ask, release,
blacklistAdditions, blacklistRemovals); //调有RM端的调度器 rScheduler
..
allocateResponse.setUpdatedNodes(updatedNodeReports);
}
//封装一个response返回
allocateResponse.setAllocatedContainers(allocation.getContainers());
allocateResponse.setCompletedContainersStatuses(appAttempt
.pullJustFinishedContainers());
allocateResponse.setResponseId(lastResponse.getResponseId() + 1);
allocateResponse.setAvailableResources(allocation.getResourceLimit());
allocateResponse.setNumClusterNodes(this.rScheduler.getNumClusterNodes());
// add preemption to the allocateResponse message (if any)
allocateResponse.setPreemptionMessage(generatePreemptionMessage(allocation));
// Adding NMTokens for allocated containers.
if (!allocation.getContainers().isEmpty()) {
allocateResponse.setNMTokens(rmContext.getNMTokenSecretManager()
.createAndGetNMTokens(app.getUser(), appAttemptId,
//FIFO Scheduler的allocate方法
...
// Update application requests
application.updateResourceRequests(ask); //将此次资源请求写入application的请求内存结构，等待nm发送心跳分配完后，写入application的分配内存结构,
//最终要更新到这样的一个内存结构 final Map<Priority, Map<String, ResourceRequest>> requests =
// new HashMap<Priority, Map<String, ResourceRequest>>();
...
return new Allocation(
application.pullNewlyAllocatedContainers(), //application内部的集合类，从分配好的内存结构里取
application.getHeadroom());
//application为FiCaSchedulerApp类
synchronized public List<Container> pullNewlyAllocatedContainers() {
List<Container> returnContainerList = new ArrayList<Container>(
newlyAllocatedContainers.size());
for (RMContainer rmContainer : newlyAllocatedContainers) { //只是从newlyAllocatedContainers里面取,newlyAllocatedContainers的赋值是NM发送心跳后调用assignContainer后赋值的
rmContainer.handle(new RMContainerEvent(rmContainer.getContainerId(),
RMContainerEventType.ACQUIRED));
returnContainerList.add(rmContainer.getContainer());
}
newlyAllocatedContainers.clear();
return returnContainerList;
}
synchronized public RMContainer allocate(NodeType type, FiCaSchedulerNode node,
Priority priority, ResourceRequest request,
Container container) {
....
// Add it to allContainers list.
newlyAllocatedContainers.add(rmContainer); //给其赋值
//FIFO scheduler类调用上面方法，该方法是NM发送心跳最终调用的方法
private int assignContainer(FiCaSchedulerNode node, FiCaSchedulerApp application,
Priority priority, int assignableContainers,
ResourceRequest request, NodeType type) {
....
}
// Create the container
Container container =
BuilderUtils.newContainer(containerId, nodeId, node.getRMNode()
.getHttpAddress(), capability, priority, containerToken);
// Allocate!
// Inform the application
RMContainer rmContainer =
application.allocate(type, node, priority, request, container);
//总结以上看到的，也就是appmaster向RM发送请求，是从当前内存结构返回资源请求，这个过程是异步的，当nm发送心跳，会根据appmaster的资源请求分配资源
//写到内存结构，等appmaster来取 (发送的资源请求，要先保存下来,资源请求的内存结构里,保存在application FiCaSchedulerApp里application.showRequests()标签：des blog http os io java ar for 2014
原文地址：http://blog.csdn.net/u011750989/article/details/39152383
