2. 程式人生 > >YARN原始碼分析之ApplicationMaster分配策略

YARN原始碼分析之ApplicationMaster分配策略

阿新 • • 發佈：2019-01-02

一次和朋友的談話中涉及到ApplicationMaster的container分配策略是什麼，我映像中是隨機分配的，但他說是根據各節點空閒資源來分配的。
之前看程式碼的時候也沒注意這塊的邏輯，既然現在有了疑惑那就去程式碼裡瞅瞅。

從MR的執行log中可以找到AM的container是在什麼時候分配的，見log

2017-04-09 03:26:17,113 INFO org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptImpl: appattempt_1491729774382_0001_000001 State change from SUBMITTED to SCHEDULED
2017
 
-04-09 03:26:17,415 INFO org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerImpl: container_1491729774382_0001_01_000001 Container Transitioned from NEW to ALLOCATED

AM container是在appattempt的狀態由SUBMITTED變為SCHEDULED時初始化的。
appattempt由SUBMITTED變為SCHEDULED狀態的處理邏輯為:

public static final class ScheduleTransition
 

    implements
    MultipleArcTransition<RMAppAttemptImpl, RMAppAttemptEvent, RMAppAttemptState> {
  @Override
  public RMAppAttemptState transition(RMAppAttemptImpl appAttempt,
      RMAppAttemptEvent event) {
    ApplicationSubmissionContext subCtx = appAttempt.submissionContext;
    if (!subCtx.getUnmanagedAM()) {
      // Need reset #containers before create new attempt, because this request
 

      // will be passed to scheduler, and scheduler will deduct the number after
      // AM container allocated
      // 設定am container的請求
      appAttempt.amReq.setNumContainers(1);
      appAttempt.amReq.setPriority(AM_CONTAINER_PRIORITY);
      // ResourceName為ANY表示任何機架上的任一機器
      appAttempt.amReq.setResourceName(ResourceRequest.ANY);
      appAttempt.amReq.setRelaxLocality(true);

      // 由排程器來分配資源
      Allocation amContainerAllocation =
          appAttempt.scheduler.allocate(appAttempt.applicationAttemptId,
              Collections.singletonList(appAttempt.amReq),
              EMPTY_CONTAINER_RELEASE_LIST, null, null);
      ...
      return RMAppAttemptState.SCHEDULED;
    } else {
      ...
    }
  }
}

首先為AM container構造container請求，其實從appAttempt.amReq.setResourceName(ResourceRequest.ANY)就可以看出am container的分配原則是隨機的，因為在建立請求時對ResourceName並沒有要求。但我們還是繼續看下程式碼以驗證下。
請求建立成功之後，由排程器來分配資源，這裡預設使用的是Capacity排程，程式碼如下:

// CapacityScheduler.java
public Allocation allocate(ApplicationAttemptId applicationAttemptId,
    List<ResourceRequest> ask, List<ContainerId> release, 
    List<String> blacklistAdditions, List<String> blacklistRemovals) {

  FiCaSchedulerApp application = getApplicationAttempt(applicationAttemptId);
  ...
  // Release containers
  releaseContainers(release, application);

  synchronized (application) {
    ...
    if (!ask.isEmpty()) {
      ...
      application.showRequests();
      // 將請求該application attempt的map中
      // Update application requests
      application.updateResourceRequests(ask);
      application.showRequests();
    }

    application.updateBlacklist(blacklistAdditions, blacklistRemovals);
    //
    return application.getAllocation(getResourceCalculator(),
                 clusterResource, getMinimumResourceCapability());
  }
}

CapacityScheduler分配請求時，呼叫application.updateResourceRequests(ask)將請求放入map中，等待nm心跳時來取。
這個application是FiCaSchedulerApp的物件，FiCaSchedulerApp其實對應的是application attempt。updateResurceRequests程式碼如下:

public synchronized void updateResourceRequests(
    List<ResourceRequest> requests) {
  if (!isStopped) {
    // AppSchedulingInfo.updateResourceRequests
    appSchedulingInfo.updateResourceRequests(requests, false);
  }
}

AppSchedulingInfo記錄了application的所有消費情況，當然也包括這個application正在執行或者已經完成的container。

synchronized public void updateResourceRequests(
    List<ResourceRequest> requests, boolean recoverPreemptedRequest) {
  // Update resource requests
  for (ResourceRequest request : requests) {
    Priority priority = request.getPriority();
    String resourceName = request.getResourceName();
    boolean updatePendingResources = false;
    ResourceRequest lastRequest = null;
    // 如果request的ResourceName是ResourceRequest.ANY
    // 只有am container是ANY？？？不應該吧
    if (resourceName.equals(ResourceRequest.ANY)) {
      ...
      // ResourceRequest.ANY才置為true？？
      updatePendingResources = true;

      // Premature optimization?
      // Assumes that we won't see more than one priority request updated
      // in one call, reasonable assumption... however, it's totally safe
      // to activate same application more than once.
      // Thus we don't need another loop ala the one in decrementOutstanding()  
      // which is needed during deactivate.
      if (request.getNumContainers() > 0) {
        activeUsersManager.activateApplication(user, applicationId);
      }
    }
    // requests是一個請求列表 map
    // 檢視requests中是否已有該優先順序的請求
    // this.requests中存放的是這個application的request
    Map<String, ResourceRequest> asks = this.requests.get(priority);
    // 沒有此優先順序的請求，則new一個map
    if (asks == null) {
      asks = new HashMap<String, ResourceRequest>();
      this.requests.put(priority, asks);
      this.priorities.add(priority);
    }
    // asks不為null，檢視asks中是否有與此請求ResourceName一樣的請求
    lastRequest = asks.get(resourceName);

    if (recoverPreemptedRequest && lastRequest != null) {
      // Increment the number of containers to 1, as it is recovering a
      // single container.
      request.setNumContainers(lastRequest.getNumContainers() + 1);
    }
    // 把原來的請求拿出賦值給lastRequest，
    // 然後將新的request將入asks中，lastRequest怎麼辦？在哪處理的？
    asks.put(resourceName, request);
    if (updatePendingResources) {

      // Similarly, deactivate application?
      if (request.getNumContainers() <= 0) {
        LOG.info("checking for deactivate... ");
        checkForDeactivation();
      }

      int lastRequestContainers = lastRequest != null ? lastRequest
          .getNumContainers() : 0;
      Resource lastRequestCapability = lastRequest != null ? lastRequest
          .getCapability() : Resources.none();
      metrics.incrPendingResources(user, request.getNumContainers(),
          request.getCapability());
      metrics.decrPendingResources(user, lastRequestContainers,
          lastRequestCapability);
    }
  }
}

updateResourceRequests主要是將請求放入requests中，等待nm心跳來取。不過這裡有點模糊，在更新requests之前，如果存在該ResourceName的請求則取出，賦值給lastRequest，然後這個lastRequest是怎麼處理的呢？不知道怎麼回事，標註下。


更新完requests之後，回到CapacityScheduler.allocate中繼續執行，return時執行application.getAllocation返回一個Allocation物件，這裡會給container建立TOKEN，這裡建立TOKEN的container是已經分配給nm的，也就是已經例項化的RMContainer，是不是說排程器在排程container時，先建立一個請求，然後從newlyAllocatedContainers列表中取出上次請求container的響應結果？

am container的請求建立好之後，等待nm心跳來取

某個nm傳送來了心跳，
程式碼如下:

// CapacityScheduler.handle  NODE_UPDATE事件
case NODE_UPDATE:
{
  NodeUpdateSchedulerEvent nodeUpdatedEvent = (NodeUpdateSchedulerEvent)event;
  RMNode node = nodeUpdatedEvent.getRMNode();
  // 更新該節點上的container的資訊
  // 對剛分配到該節點的container進行launch，已經完成的container進行狀態轉移
  nodeUpdate(node);
  if (!scheduleAsynchronously) {
    // 該節點取container請求
    allocateContainersToNode(getNode(node.getNodeID()));
  }
}

nm與CapacityScheduler心跳之後，通過nodeUpdate(node)對改節點上已有的container進行狀態更新，然後呼叫allocateContainersToNode去拉取新的container請求。

private synchronized void allocateContainersToNode(FiCaSchedulerNode node) {
  ...
  // Assign new containers...
  // 1. Check for reserved applications
  // 2. Schedule if there are no reservations
  // 如果有預留container的話先分配預留的container
  ...
  // Try to schedule more if there are no reservations to fulfill
  if (node.getReservedContainer() == null) {
    // 計算nm上是否還有空閒的資源進行分配container
    if (calculator.computeAvailableContainers(node.getAvailableResource(),
      minimumAllocation) > 0) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Trying to schedule on node: " + node.getNodeName() +
            ", available: " + node.getAvailableResource());
      }
      // 分配container
      root.assignContainers(clusterResource, node, false);
    }
  } else {
    LOG.info("Skipping scheduling since node " + node.getNodeID() + 
        " is reserved by application " + 
        node.getReservedContainer().getContainerId().getApplicationAttemptId()
        );
  }
}

排程器給這臺nm排程container時，先判斷這臺nm上是否有預留的container，如果有先對預留的container進行分配，如果沒有預留的container才呼叫root.assignContainers進行排程。
root是CSQueue物件，CSQueue是一個介面，抽象類AbstractCSQueue實現了該介面，而AbstractCSQueue又被ParentQueue和ChildQueue繼承，這裡呼叫的是ParentQueue的assignContainers，程式碼如下:

public synchronized CSAssignment assignContainers(
    Resource clusterResource, FiCaSchedulerNode node, boolean needToUnreserve) {
  CSAssignment assignment = 
      new CSAssignment(Resources.createResource(0, 0), NodeType.NODE_LOCAL);

  // if our queue cannot access this node, just return
  if (!SchedulerUtils.checkQueueAccessToNode(accessibleLabels,
      labelManager.getLabelsOnNode(node.getNodeID()))) {
    return assignment;
  }

  while (canAssign(clusterResource, node)) {
    ...
    boolean localNeedToUnreserve = false;
    Set<String> nodeLabels = labelManager.getLabelsOnNode(node.getNodeID()); 
    // Are we over maximum-capacity for this queue?
    if (!canAssignToThisQueue(clusterResource, nodeLabels)) {
      // check to see if we could if we unreserve first
      localNeedToUnreserve = assignToQueueIfUnreserve(clusterResource);
      if (!localNeedToUnreserve) {
        break;
      }
    }
    // Schedule
    CSAssignment assignedToChild = 
        assignContainersToChildQueues(clusterResource, node, localNeedToUnreserve | needToUnreserve);
    assignment.setType(assignedToChild.getType());
    ...
    // Do not assign more than one container if this isn't the root queue
    // or if we've already assigned an off-switch container
    if (!rootQueue || assignment.getType() == NodeType.OFF_SWITCH) {
      if (LOG.isDebugEnabled()) {
        if (rootQueue && assignment.getType() == NodeType.OFF_SWITCH) {
          LOG.debug("Not assigning more than one off-switch container," +
              " assignments so far: " + assignment);
        }
      }
      break;
    }
  } 

  return assignment;
}

分配時，先判斷此佇列是否可以訪問該nm，然後判斷是否可以訪問該nm上的label，都判斷通過之後呼叫assignContainersToChildQueues進行分配，

private synchronized CSAssignment assignContainersToChildQueues(Resource cluster, 
    FiCaSchedulerNode node, boolean needToUnreserve) {
  ...
  // Try to assign to most 'under-served' sub-queue
  for (Iterator<CSQueue> iter=childQueues.iterator(); iter.hasNext();) {
    CSQueue childQueue = iter.next();
    ...
    assignment = childQueue.assignContainers(cluster, node, needToUnreserve);
    ...
    // If we do assign, remove the queue and re-insert in-order to re-sort
    if (Resources.greaterThan(
            resourceCalculator, cluster, 
            assignment.getResource(), Resources.none())) {
      // Remove and re-insert to sort
      iter.remove();
      LOG.info("Re-sorting assigned queue: " + childQueue.getQueuePath() + 
          " stats: " + childQueue);
      childQueues.add(childQueue);
      if (LOG.isDebugEnabled()) {
        printChildQueues();
      }
      break;
    }
  }

  return assignment;
}

assignContainersToChildQueues呼叫ChildQueue的assignContainer進行分配，分配之後要講改childQueue從佇列中remove掉，然後重新插入到佇列中，進行排序。
childQueue.assignContainers如下:

public synchronized CSAssignment assignContainers(Resource clusterResource,
    FiCaSchedulerNode node, boolean needToUnreserve) {
  ...
  // if our queue cannot access this node, just return
  if (!SchedulerUtils.checkQueueAccessToNode(accessibleLabels,
      labelManager.getLabelsOnNode(node.getNodeID()))) {
    return NULL_ASSIGNMENT;
  }

  // Check for reserved resources
  RMContainer reservedContainer = node.getReservedContainer();
  if (reservedContainer != null) {
    FiCaSchedulerApp application = 
        getApplication(reservedContainer.getApplicationAttemptId());
    synchronized (application) {
      return assignReservedContainer(application, node, reservedContainer,
          clusterResource);
    }
  }

  // Try to assign containers to applications in order
  for (FiCaSchedulerApp application : activeApplications) {
    ...
    // 加鎖
    synchronized (application) {
      // Check if this resource is on the blacklist
      if (SchedulerAppUtils.isBlacklisted(application, node, LOG)) {
        continue;
      }

      // Schedule in priority order
      for (Priority priority : application.getPriorities()) {
        // 為什麼是ANY？
        // 如果當前application中的request中沒有ANY就不分配？
        // 想辦法debug試一下
        ResourceRequest anyRequest =
            application.getResourceRequest(priority, ResourceRequest.ANY);
        if (null == anyRequest) {
          continue;
        }

        // Required resource
        Resource required = anyRequest.getCapability();

        // Do we need containers at this 'priority'?
        if (application.getTotalRequiredResources(priority) <= 0) {
          continue;
        }
        if (!this.reservationsContinueLooking) {
          if (!needContainers(application, priority, required)) {
            if (LOG.isDebugEnabled()) {
              LOG.debug("doesn't need containers based on reservation algo!");
            }
            continue;
          }
        }

        Set<String> requestedNodeLabels =
            getRequestLabelSetByExpression(anyRequest
                .getNodeLabelExpression());

        // Compute user-limit & set headroom
        // Note: We compute both user-limit & headroom with the highest 
        //       priority request as the target. 
        //       This works since we never assign lower priority requests
        //       before all higher priority ones are serviced.
        Resource userLimit = 
            computeUserLimitAndSetHeadroom(application, clusterResource, 
                required, requestedNodeLabels);          

        // Check queue max-capacity limit
        if (!canAssignToThisQueue(clusterResource, required,
            labelManager.getLabelsOnNode(node.getNodeID()), application, true)) {
          return NULL_ASSIGNMENT;
        }

        // Check user limit
        if (!assignToUser(clusterResource, application.getUser(), userLimit,
            application, true, requestedNodeLabels)) {
          break;
        }

        // Inform the application it is about to get a scheduling opportunity
        // 這又是什麼鬼？增加排程的機會？
        application.addSchedulingOpportunity(priority);

        // Try to schedule
        // 開始排程
        CSAssignment assignment =  
          assignContainersOnNode(clusterResource, node, application, priority, 
              null, needToUnreserve);

        // Did the application skip this node?
        if (assignment.getSkipped()) {
          // Don't count 'skipped nodes' as a scheduling opportunity!
          application.subtractSchedulingOpportunity(priority);
          continue;
        }

        // Did we schedule or reserve a container?
        Resource assigned = assignment.getResource();
        if (Resources.greaterThan(
            resourceCalculator, clusterResource, assigned, Resources.none())) {

          // Book-keeping 
          // Note: Update headroom to account for current allocation too...
          allocateResource(clusterResource, application, assigned,
              labelManager.getLabelsOnNode(node.getNodeID()));

          // Don't reset scheduling opportunities for non-local assignments
          // otherwise the app will be delayed for each non-local assignment.
          // This helps apps with many off-cluster requests schedule faster.
          if (assignment.getType() != NodeType.OFF_SWITCH) {
            if (LOG.isDebugEnabled()) {
              LOG.debug("Resetting scheduling opportunities");
            }
            application.resetSchedulingOpportunities(priority);
          }

          // Done
          return assignment;
        } else {
          // Do not assign out of order w.r.t priorities
          break;
        }
      }
    }

    if(LOG.isDebugEnabled()) {
      LOG.debug("post-assignContainers for application "
        + application.getApplicationId());
    }
    application.showRequests();
  }

  return NULL_ASSIGNMENT;
}

LeafQueue.assignContainers會從遍歷當前佇列中正在執行的application的container請求，通過一系列的邏輯之後呼叫assignContainersOnNode進行排程

private CSAssignment assignContainersOnNode(Resource clusterResource,
    FiCaSchedulerNode node, FiCaSchedulerApp application, Priority priority,
    RMContainer reservedContainer, boolean needToUnreserve) {
  Resource assigned = Resources.none();
  // 如果ResourceName是NODE_LOCAL
  ResourceRequest nodeLocalResourceRequest =
      application.getResourceRequest(priority, node.getNodeName());
  if (nodeLocalResourceRequest != null) {
    assigned = 
        assignNodeLocalContainers(clusterResource, nodeLocalResourceRequest, 
            node, application, priority, reservedContainer, needToUnreserve); 
    if (Resources.greaterThan(resourceCalculator, clusterResource, 
        assigned, Resources.none())) {
      return new CSAssignment(assigned, NodeType.NODE_LOCAL);
    }
  }
  // 如果ResourceName是Rack-local
  ResourceRequest rackLocalResourceRequest =
      application.getResourceRequest(priority, node.getRackName());
  if (rackLocalResourceRequest != null) {
    if (!rackLocalResourceRequest.getRelaxLocality()) {
      return SKIP_ASSIGNMENT;
    }

    assigned = 
        assignRackLocalContainers(clusterResource, rackLocalResourceRequest, 
            node, application, priority, reservedContainer, needToUnreserve);
    if (Resources.greaterThan(resourceCalculator, clusterResource, 
        assigned, Resources.none())) {
      return new CSAssignment(assigned, NodeType.RACK_LOCAL);
    }
  }

  // 如果ResourceName是Off-switch，也就是ANY
  ResourceRequest offSwitchResourceRequest =
      application.getResourceRequest(priority, ResourceRequest.ANY);
  if (offSwitchResourceRequest != null) {
    if (!offSwitchResourceRequest.getRelaxLocality()) {
      return SKIP_ASSIGNMENT;
    }

    return new CSAssignment(
        assignOffSwitchContainers(clusterResource, offSwitchResourceRequest,
            node, application, priority, reservedContainer, needToUnreserve), 
            NodeType.OFF_SWITCH);
  }
  return SKIP_ASSIGNMENT;
}

assignContainersOnNode會根據請求中資源的型別進行不同的邏輯處理，由於am container中ResourceRequest為ANY，所以這裡只關注下Off-switch的處理邏輯，程式碼如下：

private Resource assignOffSwitchContainers(
    Resource clusterResource, ResourceRequest offSwitchResourceRequest,
    FiCaSchedulerNode node, FiCaSchedulerApp application, Priority priority, 
    RMContainer reservedContainer, boolean needToUnreserve) {
  if (canAssign(application, priority, node, NodeType.OFF_SWITCH, 
      reservedContainer)) {
    return assignContainer(clusterResource, node, application, priority,
        offSwitchResourceRequest, NodeType.OFF_SWITCH, reservedContainer,
        needToUnreserve);
  }

  return Resources.none();
}

assignOffSwitchContainers又呼叫了assignContainer，繼續跟蹤

private Resource assignContainer(Resource clusterResource, FiCaSchedulerNode node, 
    FiCaSchedulerApp application, Priority priority, 
    ResourceRequest request, NodeType type, RMContainer rmContainer,
    boolean needToUnreserve) {
  ...
  // container的資源大小  
  Resource capability = request.getCapability();
  // 節點可用的資源大小
  Resource available = node.getAvailableResource();
  // 節點總共資源大小
  Resource totalResource = node.getTotalResource();
  // 判斷請求的資源是否超過了節點的總量
  if (!Resources.fitsIn(capability, totalResource)) {
    LOG.warn("Node : " + node.getNodeID()
        + " does not have sufficient resource for request : " + request
        + " node total capability : " + node.getTotalResource());
    return Resources.none();
  }
  assert Resources.greaterThan(
      resourceCalculator, clusterResource, available, Resources.none());

  // Create the container if necessary
  // 生成containerId
  Container container = 
      getContainer(rmContainer, application, node, capability, priority);
  ...
  // 先判斷是否可以分配預留的container，
  // 可以分配正常的container時，才去判斷空閒的資源是否可以分配
  // Can we allocate a container on this node?
  int availableContainers = 
      resourceCalculator.computeAvailableContainers(available, capability);
  if (availableContainers > 0) {
    // Allocate...
    ...
    // Inform the application
    RMContainer allocatedContainer = 
        application.allocate(type, node, priority, request, container);

    // Does the application need this resource?
    if (allocatedContainer == null) {
      return Resources.none();
    }
    // 通知node進行分配，將container放入launchedContainers map中
    // Inform the node
    node.allocateContainer(allocatedContainer);

    LOG.info("assignedContainer" +
        " application attempt=" + application.getApplicationAttemptId() +
        " container=" + container + 
        " queue=" + this + 
        " clusterResource=" + clusterResource);

    return container.getResource();
  } else {
    // if we are allowed to allocate but this node doesn't have space, reserve it or
    // if this was an already a reserved container, reserve it again
    ...
    return Resources.none();
  }
}

assignContainer首先判斷container請求的資源是否超過了節點的總資源量，如果沒有超過呼叫getContainer檢視當前節點上是否有預留的container，沒有則createContainer，生成containerId。containerId生成之後，去判斷當前節點上的空閒資源能否夠分配，如果可以的話就呼叫application.allocate進行分配，application是FiCaSchedulerApp的物件。最後將container放入launchedContainers中，隨後會心跳返回給node。allocate程式碼如下:

synchronized public RMContainer allocate(NodeType type, FiCaSchedulerNode node,
    Priority priority, ResourceRequest request, 
    Container container) {
  ...
  // container在RM端稱為RMcontainer
  // Create RMContainer
  RMContainer rmContainer = new RMContainerImpl(container, this
      .getApplicationAttemptId(), node.getNodeID(),
      appSchedulingInfo.getUser(), this.rmContext);

  // Add it to allContainers list.
  // 將生成的container放入allContainers list
  // 排程器在排程的時候從中取出container
  newlyAllocatedContainers.add(rmContainer);
  liveContainers.put(container.getId(), rmContainer);    

  // Update consumption and track allocations
  List<ResourceRequest> resourceRequestList = appSchedulingInfo.allocate(
      type, node, priority, request, container);
  Resources.addTo(currentConsumption, container.getResource());

  // Update resource requests related to "request" and store in RMContainer 
  ((RMContainerImpl)rmContainer).setResourceRequests(resourceRequestList);

  // Inform the container
  // 時間排程器來通知container已經準備好，觸發container狀態機
  rmContainer.handle(
      new RMContainerEvent(container.getId(), RMContainerEventType.START));
  ...
  return rmContainer;
}

allocate建立一個RMContainer，並將其放入allContainers列表newlyAllocatedContainers中，排程器從中取出container分配給node。

總結

大致的流程順著程式碼理解的差不多了，但一些細節還是沒有搞太清楚，隨後詳細debug下，在更新吧。
說下我目前的理解，排程器首先建立一個container請求，並檢視newlyAllocatedContainers中是否有可排程的container，如果有則建立該container的TOKEN。然後nm來進行心跳的時候，從requests中取出對應的請求進行例項化，隨後再放入newlyAllocatedContainers列表中，等待排程。

相關推薦

YARN原始碼分析之ApplicationMaster分配策略

一次和朋友的談話中涉及到ApplicationMaster的container分配策略是什麼，我映像中是隨機分配的，但他說是根據各節點空閒資源來分配的。 之前看程式碼的時候也沒注意這塊的邏輯，既然現在有了疑惑那就去程式碼裡瞅瞅。 從MR的執行log中

elasticsearch原始碼分析之分片分配（十）

分片 什麼是分片 分片是把索引資料切分成多個小的索引塊，這些小的索引塊能夠分發到同一個叢集中的不同節點。在檢索時，檢索結果是該索引每個分片上檢索結果的合併。類似於資料庫的分庫分表。 為什麼分片 1、這樣可以提高讀寫效能，實現負載均衡。 2、副本容易

Yarn原始碼分析之旅---總體架構---概述與總體架構

歡迎大家討論，我也是接觸時間不長，有問題歡迎大家指正。歡迎轉載，轉載請註明出處 Haddoop 1.0的不足與Hadoop2.0的產生         學習和研究過Hadoop1.0的人都應該知道，在Hadoop1.0中，使用了Master\Slave的架構模式，jobTr

Yarn原始碼分析之MRAppMaster：作業執行方式Local、Uber、Non-Uber

        基於作業大小因素，MRAppMaster提供了三種作業執行方式：本地Local模式、Uber模式、Non-Uber模式。其中，        1、本地Local模式：通常用於除錯；        2、Uber模式：為降低小作業延遲而設計的一種模式，所有任務，不

Yarn原始碼分析之MapReduce作業中任務Task排程整體流程（一）

        v2版本的MapReduce作業中，作業JOB_SETUP_COMPLETED事件的發生，即作業SETUP階段完成事件，會觸發作業由SETUP狀態轉換到RUNNING狀態，而作業狀態轉換中涉及作業資訊的處理，是由SetupCompletedTransition

Spark原始碼分析之Spark Shell（上）

https://www.cnblogs.com/xing901022/p/6412619.html 文中分析的spark版本為apache的spark-2.1.0-bin-hadoop2.7。 bin目錄結構： -rwxr-xr-x. 1 bigdata bigdata 1089 Dec

Netty 原始碼分析之拆包器的奧祕

為什麼要粘包拆包 為什麼要粘包 首先你得了解一下TCP/IP協議，在使用者資料量非常小的情況下，極端情況下，一個位元組，該TCP資料包的有效載荷非常低，傳遞100位元組的資料，需要100次TCP傳送，100次ACK，在應用及時性要求不高的情況下，將這100個有效資料拼接成一個數據包，那會縮短到一個TCP資

Android原始碼分析之為什麼在onCreate() 和 onResume() 獲取不到 View 的寬高

轉載自：https://www.jianshu.com/p/d7ab114ac1f7 先來看一段很熟悉的程式碼，可能在最開始接觸安卓的時候，大部分人都寫過的一段程式碼；即嘗試在 onCreate() 和 onResume() 方法中去獲取某個 View 的寬高資訊： 但是列印輸出後，我們會發

netty原始碼分析之服務端啟動

ServerBootstrap與Bootstrap分別是netty中服務端與客戶端的引導類，主要負責服務端與客戶端初始化、配置及啟動引導等工作，接下來我們就通過netty原始碼中的示例對ServerBootstrap與Bootstrap的原始碼進行一個簡單的分析。首先我們知道這兩個類都繼承自AbstractB

SNMP原始碼分析之（一）配置檔案部分

snmpd.conf想必不陌生。在程序啟動過程中會去讀取配置檔案中各個配置。其中幾個引數需要先知道是幹什麼的：   token：配置檔案的每行的開頭，例如 group MyROGroup v1 readSec 這行token的引數是group。  

【kubernetes/k8s原始碼分析】kubelet原始碼分析之cdvisor原始碼分析

  資料流 UnsecuredDependencies -> run   1. cadvisor.New初始化 if kubeDeps.CAdvisorInterface == nil { imageFsInfoProvider := cadv

【kubernetes/k8s原始碼分析】kubelet原始碼分析之容器網路初始化原始碼分析

一. 網路基礎   1.1 網路名稱空間的操作 建立網路名稱空間： ip netns add 名稱空間內執行命令： ip netns exec 進入名稱空間： ip netns exec bash   1.2 bridge-nf-c

【kubernetes/k8s原始碼分析】kubelet原始碼分析之資源上報

0. 資料流   路徑： pkg/kubelet/kubelet.go   Run函式（） ->   syncNodeStatus ()  ->   registerWithAPIServer() ->

【kubernetes/k8s原始碼分析】kubelet原始碼分析之啟動容器

主要是呼叫runtime，這裡預設為docker 0. 資料流 NewMainKubelet（cmd/kubelet/app/server.go） -> NewKubeGenericRuntimeManager(pkg/kubelet/kuberuntime/kuberuntime

Android系統原始碼分析之-ContentProvider

距離上一次寫部落格已經半年多了，這半年發生了很多事情，也有了很多感觸，最主要是改變了忙碌了工作，更加重視身體的健康，為此也把工作地點從深圳這個一線城市換到了珠海，工作相對沒有那麼累，身體感覺也好了很多。所以在工作完成之餘，也有了更多的時間來自我學習和提高，後續會用更多時間來寫更多實用的東西，幫助我們理解

Vue 原始碼分析之proxy代理

Vue 原始碼分析之proxy代理 當我們在使用Vue進行資料設定時，通常初始化格式為: let data = { age: 12, name: 'yang' } // 例項化Vue物件 let vm = new Vue({ data })

Qt原始碼分析之事件分發器QEventDispatcherWin32

分析Qt原始碼一則想自己在開發學習中有積累，同時自己也一直有一種理念，使用她那麼就更深入的認識她。 如果有分析不正確的，還煩請各位看官指正。 事件分發器建立 在QCoreApplication建構函式中 if (!QCoreApplicationPrivate

lodash原始碼分析之isArguments

lodash原始碼分析之isArguments 有人命中註定要過平庸的生活，默默無聞，因為他們經歷了痛苦或不幸；有人卻故意這樣做，那是因為他們得到的幸福超過了他們的承受能力。 ——卡爾維諾《煙雲》 本文為讀 lodash 原始碼的第二十一篇，後續文章會更新到這個倉庫中，歡迎 star：poc

Netty原始碼分析之LengthFieldBasedFrameDecoder

拆包的原理 關於拆包原理的上一篇博文 netty原始碼分析之拆包器的奧祕 中已詳細闡述，這裡簡單總結下：netty的拆包過程和自己寫手工拆包並沒有什麼不同，都是將位元組累加到一個容器裡面，判斷當前累加的位元組資料是否達到了一個包的大小，達到一個包大小就拆開，進而傳遞到上層業務解碼handler 之所以ne

illuminate/routing 原始碼分析之註冊路由

我們知道，在 Laravel 世界裡，外界傳進來一個 Request 時，會被 Kernel 處理並返回給外界一個 Response。Kernel 在處理 Request 時，會呼叫 illuminate/routing 包提供的路由功能，來根據當前的 Request，轉發到對應的執行邏輯（執行邏輯的形式可以