Overview

• ...

YARN Architecture

• The fundamental idea of YARN is to split up the functionalities of resource management and job scheduling/monitoring into separate daemons. [基本思想是將資源管理和任務調度/監控分開。]
• The idea is to have a global ResourceManager(RM) and per-application ApplicationMaster(AM). The application is either a single job or a DAG of jobs. [思路是通過一個全局的資源管理器和一個per-app的應用管理器。]
• AM: AM is a framework specific library and is tasked with negotiating resources from the RM and working with the NM(s) to execute and monitor the tasks. [AM是由框架指定的，任務是與RM協調資源，並監控NM(s)執行tasks。]
• RM: RM has two main components: Scheduler and ApplicationsManager. [Attention: RM有兩個組件，其中Scheduler完全就只是負責資源的分配；ApplicationsManager則負責接受application，選取ApplicationMaster，監控重啟AM。]

• The Scheduler is responsible for allocating resources to the various running applications subject to familiar constraints of capacities, queues etc. The Scheduler is pure scheduler in the sense that it performs no monitoring or tracking of status for the application. Also, it offers no guarantees about restarting failed tasks. The Shceduler performs its scheduling function based on the resource requirements of the applications; it does so based on the abstract notion of a resource Container which incorporates elements such as memory, cpu, disk, network etc. The scheduler has a pluggable policy which is responsible for partitioning the cluster resources among the various queues, applications etc. The current schedulers such as the CapacityScheduler and the FairScheduler would be some examples of plug-ins.[Scheduler負責給運行的applications分配資源。Scheduler不會執行任何的監控或狀態追蹤。Scheduler基於application的資源需求進行調度，具體包括內存、cpu、硬盤、網絡等。]
• The ApplicationsManager is responsible for accepting job-submissions, negotiating the first container of executing the application specific ApplicationMaster and provides the service for restarting the ApplicationMaster container on failure. [RM中的ApplicationsMaster負責接受任務提交，協調第一個應用執行容器作為ApplicationMaster，並且負責ApplicationMaster失敗時的重啟。]

Capacity Scheculer

• Capacity Scheduler: a pluggable scheduler for Hadoop which allows for multuple-tenants to securely share a large cluster such that their applications are allocated resources in a timely manner under constraints of allocated capacities.
• TBD...

YARN資源調度策略

什麽資源？

• 一般來說，在一個分布式、多用戶的系統中，我們所指的資源通常是硬件資源，包括CPU使用、內存使用、磁盤用量、IO、網絡流量等。這些是比較粗粒度的，也可以考慮更高抽象層次的TPS/請求數等等。
• YARN的資源抽象比較簡單，只有兩種：內存和CPU。

基本概念

Container

• Container是RM分配資源的基本單位。RM負責接收用戶的資源請求，並為其分配Container，而NM負責啟動Container並監控資源使用。
• Container的作用不只於資源分配，還用於資源隔離。更進一步，client可以通過Container來要求只在特定節點上分配，從而保證了計算本地性。

調度器與隊列

• 在YARN中，調度器是一個可插拔的組件，常見的有FIFO，CapacityScheduler，FairScheduler，可通過配置文件選擇不同的調度器。
• 在RM端，根據不同的調度器，所有資源被分為一個或多個隊列，每個隊列包含一定量的資源。
• 因而，調度器有兩個主要功能：
  • 決定如何劃分隊列；
  • 決定如何分配資源。
• 此外，調度器還有一些其他的，諸如ACL、搶占、延遲調度等特性。

事件驅動

• YARN實現了一套基於狀態機的事件驅動機制：很多對象內部都有一個預先定義好的有限狀態機，相應的事件會觸發狀態轉換，狀態轉換的過程中會觸發預先定義的鉤子，鉤子執行的過程中又生成新的事件，繼續狀態轉換。
• 幾個角色：
  • Dispatcher —— 用於分發事件，一般是異步的。內部用一個BlockingQueue暫存所有事件。
  • Event —— 事件類型。
  • Handler —— 事件的消費者。每個消費者只handle特定的事件，所有Handler要在Dispatcher上註冊。

pull-based

• AM通過心跳向RM申請資源，但申請的資源不能馬上拿到，而是要再經過若幹次心跳才能拿到。這是一種pull-based模型。
• AM通過RPC協議ApplicationMasterProtocol與RM通信。這個協議在服務器端會調用YrnScheduler的allocate方法（所有調度器都必須實現YarnScheduler接口）。allocate方法有兩個作用：1. 申請、釋放資源；2. 表示AM是否存活(心跳)。

常見調度器

FIFO

• 最簡單的默認調度器，只有一個隊列，所有用戶共享。
• 先到先得，因此很容易出現一個用戶占滿集群所有資源的情況。
• 可以設置ACL(訪問控制列表)，但不能設置各個用戶的優先級。

CapacityScheduler

• 在FIFO的支持上，增加多用戶支持。
• 最大化集群吞吐量和利用率。
• 基本思想：每個用戶可以使用特定量的資源，但集群空閑時也可以使用整個集群(因而在單用戶時和FIFO差不多。這種設計主要是為了提高集群利用率。)
• 劃分隊列：
  • 劃分隊列使用xml文件配置，每個隊列可以使用特定百分比的資源。
  • 隊列可以是樹狀結構，子隊列之和不能超過父隊列。
• CapacityScheduler一個比較重要的問題就是百分比如何計算。默認的算法是DefaultResourceCalculator類的ratio方法，只考慮了內存。
• 以上，CapacityScheduler的優點就是靈活，集群的利用率高；缺點也是由其靈活性造成的，因為CapacityScheduler不支持搶占式調度，必須等上一個任務主動釋放資源。

FairScheduler

• 優先保證“公平”的調度器，每個用戶只有特定數量的資源可以用，即使集群很空閑。
• 使用xml文件配置，每個隊列可以使用特定數量的內存和CPU。
• 隊列是樹狀結構，只有葉子結點能提交任務。
• 優缺點：穩定，管理方便，運維成本低，相對CapacityScheduler犧牲了靈活性，整體資源利用率不高。

RM REST API‘s

Overview

• The RM REST API‘s allow the user to get information about the cluster - status on the cluster, metrics on the cluster, scheduler informtion, information about nodes in the cluster, and information about applications on the cluster.

Cluster Information API

• The cluster information resource provides overall information about the cluster.
• demo usage:

• public static void getClusterInfo() throws Exception{
          String url = "http://10.3.242.99:8088/ws/v1/cluster";
          InputStream response = new URL(url).openStream();
  
          Scanner scanner = new Scanner(response);
          String responseBody = scanner.useDelimiter("\\A").next();
          System.out.println(responseBody);
  
      }
  

  返回Json：

  {
      "clusterInfo": {
          "id": 1497756603270,
          "startedOn": 1497756603270,
          "state": "STARTED",
          "haState": "ACTIVE",
          "rmStateStoreName": "org.apache.hadoop.yarn.server.resourcemanager.recovery.NullRMStateStore",
          "resourceManagerVersion": "2.6.2",
          "resourceManagerBuildVersion": "2.6.2 from 0cfd050febe4a30b1ee1551dcc527589509fb681 by jenkins source checksum d07deb9ef36deb791d0e2451db849d",
          "resourceManagerVersionBuiltOn": "2015-10-22T00:49Z",
          "hadoopVersion": "2.6.2",
          "hadoopBuildVersion": "2.6.2 from 0cfd050febe4a30b1ee1551dcc527589509fb681 by jenkins source checksum f9ebb94bf5bf9bec892825ede28baca",
          "hadoopVersionBuiltOn": "2015-10-22T00:42Z",
          "haZooKeeperConnectionState": "ResourceManager HA is not enabled."
      }
  }

  【TBD：haStatus，關於RM的HA。】

Cluster Metrics API

• The cluster metrics resource provides some overall metrics about the cluster.
• demo usage:
  

  public static void getClusterMetrics() throws Exception{
          String url = "http://10.3.242.99:8088/ws/v1/cluster/metrics";
          InputStream response = new URL(url).openStream();
  
          Scanner scanner = new Scanner(response);
          String responseBody = scanner.useDelimiter("\\A").next();
          System.out.println(responseBody);
      }
  

  return json:

  {
      "clusterMetrics": {
          "appsSubmitted": 0,
          "appsCompleted": 0,
          "appsPending": 0,
          "appsRunning": 0,
          "appsFailed": 0,
          "appsKilled": 0,
          "reservedMB": 0,
          "availableMB": 32768,
          "allocatedMB": 0,
          "reservedVirtualCores": 0,
          "availableVirtualCores": 8,
          "allocatedVirtualCores": 0,
          "containersAllocated": 0,
          "containersReserved": 0,
          "containersPending": 0,
          "totalMB": 32768,
          "totalVirtualCores": 8,
          "totalNodes": 1,
          "lostNodes": 0,
          "unhealthyNodes": 0,
          "decommissionedNodes": 0,
          "rebootedNodes": 0,
          "activeNodes": 1
      }
  } 

Cluster Scheduler API

• A scheduler resource contains information about the current scheduler configured in a cluster. It currently supports both the Fifo and Capacity Scheduler.
• You will get different information depending on which scheduler is configured so be sure to look at the type information.
• demo usage:
  

  String clusterSchedulerUrl = "http://10.3.242.99:8088/ws/v1/cluster/scheduler";
  YarnRestAPI.httpGet(clusterSchedulerUrl);
  

  return json:
  
  

  {
      "scheduler": {
          "schedulerInfo": {
              "type": "capacityScheduler",
              "capacity": 100,
              "usedCapacity": 0,
              "maxCapacity": 100,
              "queueName": "root",
              "queues": {
                  "queue": [
                      {
                          "type": "capacitySchedulerLeafQueueInfo",
                          "capacity": 100,
                          "usedCapacity": 0,
                          "maxCapacity": 100,
                          "absoluteCapacity": 100,
                          "absoluteMaxCapacity": 100,
                          "absoluteUsedCapacity": 0,
                          "numApplications": 0,
                          "queueName": "default",
                          "state": "RUNNING",
                          "resourcesUsed": {
                              "memory": 0,
                              "vCores": 0
                          },
                          "hideReservationQueues": false,
                          "nodeLabels": [
                              "*"
                          ],
                          "numActiveApplications": 0,
                          "numPendingApplications": 0,
                          "numContainers": 0,
                          "maxApplications": 10000,
                          "maxApplicationsPerUser": 10000,
                          "userLimit": 100,
                          "users": null,
                          "userLimitFactor": 1,
                          "aMResourceLimit": {
                              "memory": 4096,
                              "vCores": 1
                          },
                          "userAMResourceLimit": {
                              "memory": 4096,
                              "vCores": 1
                          }
                      }
                  ]
              }
          }
      }
  }
  

  從返回的json可以看到，使用的capacityScheduler，其內有一個queues數組。

Cluster Applications API

• With the Applications API, you can obtain a collection of resources, each of which represents an application. When you run a GET operation on this resource, you obtain a collection of Application Objects.
• demo usage:
  

  String clusterAppUrl = "http://10.3.242.99:8088/ws/v1/cluster/apps";
  YarnRestAPI.httpGet(clusterAppUrl);

　　　return json:
　　　　

{
    "apps": {
        "app": [
            {
                "id": "application_1497756603270_0002",
                "user": "root",
                "name": "word count",
                "queue": "default",
                "state": "FINISHED",
                "finalStatus": "SUCCEEDED",
                "progress": 100,
                "trackingUI": "History",
                "trackingUrl": "http://host99:8088/proxy/application_1497756603270_0002/",
                "diagnostics": "",
                "clusterId": 1497756603270,
                "applicationType": "MAPREDUCE",
                "applicationTags": "",
                "startedTime": 1497784937984,
                "finishedTime": 1497784954421,
                "elapsedTime": 16437,
                "amContainerLogs": "http://host99:8042/node/containerlogs/container_1497756603270_0002_01_000001/root",
                "amHostHttpAddress": "host99:8042",
                "allocatedMB": 2048,
                "allocatedVCores": 1,
                "runningContainers": 1,
                "memorySeconds": 239770,
                "vcoreSeconds": 29,
                "preemptedResourceMB": 0,
                "preemptedResourceVCores": 0,
                "numNonAMContainerPreempted": 0,
                "numAMContainerPreempted": 0
            },
            {
                "id": "application_1497756603270_0001",
                "user": "root",
                "name": "word count",
                "queue": "default",
                "state": "FINISHED",
                "finalStatus": "SUCCEEDED",
                "progress": 100,
                "trackingUI": "History",
                "trackingUrl": "http://host99:8088/proxy/application_1497756603270_0001/",
                "diagnostics": "",
                "clusterId": 1497756603270,
                "applicationType": "MAPREDUCE",
                "applicationTags": "",
                "startedTime": 1497784895511,
                "finishedTime": 1497784913807,
                "elapsedTime": 18296,
                "amContainerLogs": "http://host99:8042/node/containerlogs/container_1497756603270_0001_01_000001/root",
                "amHostHttpAddress": "host99:8042",
                "allocatedMB": -1,
                "allocatedVCores": -1,
                "runningContainers": -1,
                "memorySeconds": 255182,
                "vcoreSeconds": 34,
                "preemptedResourceMB": 0,
                "preemptedResourceVCores": 0,
                "numNonAMContainerPreempted": 0,
                "numAMContainerPreempted": 0
            }
        ]
    }
}

Query Parameters Supported

• Multiple parameters can be specified for GET operations.
• The started and finished times have a begin and end parameter to allow you to specify ranges. For example, you can request all applications that started between 1:00am and 2:00pm on 12/19/2011 with startedTimeBegin=1324256400&startedTimeEnd=1324303200.

Cluster Application Statistics API

• With the Application Statistics API, you can obtain a collection of triples, each of which contains the application type, the appplication state and the number of applications of this type and this state in RM context.
• demo usage:
  

  String appMetricsUrl = "http://10.3.242.99:8088/ws/v1/cluster/appstatistics";
  YarnRestAPI.httpGet(appMetricsUrl);
  

  return json:
  
  

  {
      "appStatInfo": {
          "statItem": [
              {
                  "state": "RUNNING",
                  "type": "*",
                  "count": 0
              },
              {
                  "state": "ACCEPTED",
                  "type": "*",
                  "count": 0
              },
              {
                  "state": "NEW_SAVING",
                  "type": "*",
                  "count": 0
              },
              {
                  "state": "NEW",
                  "type": "*",
                  "count": 0
              },
              {
                  "state": "KILLED",
                  "type": "*",
                  "count": 0
              },
              {
                  "state": "FINISHED",
                  "type": "*",
                  "count": 2
              },
              {
                  "state": "FAILED",
                  "type": "*",
                  "count": 0
              },
              {
                  "state": "SUBMITTED",
                  "type": "*",
                  "count": 0
              }
          ]
      }
  }

• 可以支持的查詢參數有：states & applicationTypes

Cluster Application API

• An application resource contains information about a particular application that was submitted to a cluster.
• demo usage:
  

  String appUrl = "http://10.3.242.99:8088/ws/v1/cluster/apps/application_1497756603270_0002";
  YarnRestAPI.httpGet(appUrl);
  

  return json:

  {
      "app": {
          "id": "application_1497756603270_0002",
          "user": "root",
          "name": "word count",
          "queue": "default",
          "state": "FINISHED",
          "finalStatus": "SUCCEEDED",
          "progress": 100,
          "trackingUI": "History",
          "trackingUrl": "http://host99:8088/proxy/application_1497756603270_0002/",
          "diagnostics": "",
          "clusterId": 1497756603270,
          "applicationType": "MAPREDUCE",
          "applicationTags": "",
          "startedTime": 1497784937984,
          "finishedTime": 1497784954421,
          "elapsedTime": 16437,
          "amContainerLogs": "http://host99:8042/node/containerlogs/container_1497756603270_0002_01_000001/root",
          "amHostHttpAddress": "host99:8042",
          "allocatedMB": -1,
          "allocatedVCores": -1,
          "runningContainers": -1,
          "memorySeconds": 241431,
          "vcoreSeconds": 30,
          "preemptedResourceMB": 0,
          "preemptedResourceVCores": 0,
          "numNonAMContainerPreempted": 0,
          "numAMContainerPreempted": 0
      }
  }

Cluster Application Attempts API

• TBD...

FYI

• RM REST API‘s

<YaRN><Official doc><RM REST API's>