PerfectHadoop: YARN 资源管理器

本项目提供了YARN资源管理器的Swift类封装：

• YARNResourceManager(): 用于访问YARN的集群信息，包括簇及其性能参数、调度器和程序提交情况。

连接到 YARN 资源管理器

只需要初始化YARNResourceManager()对象并提供必要参数即可连接到YARN资源管理器：

// 以下连接仅支持基本操作
let yarn = YARNResourceManager(host: "yarn.somehadoopdomain.com", port: 8088)

比如提供一个有效的用户名：

// 增加用户名
let yarn = YARNResourceManager(host: "yarn.somehadoopdomain.com", port: 8088, user: "your user name")

用户身份验证

如果您的服务器在使用 Kerberos 验证机制，请增加参数如下：

// 将认证机制设置为KRB5：
let yarn = YARNResourceManager(host: "yarn.somehadoopdomain.com", port: 8088, user: "username", auth: .krb5)

YARNResourceManager 资源管理器对象数据结构

获取服务器基本信息

调用 checkClusterInfo() 方法可以获得服务器基本信息，返回结果为一个 ClusterInfo 数据结构：

guard let i = try yarn.checkClusterInfo() else {
    // 出错了
}//end guard
print(i.startedOn)
print(i.state)
print(i.hadoopVersion)
print(i.resourceManagerVersion)

调用成功后checkClusterInfo() 会返回 ClusterInfo 集群对象信息，如下所述：

ClusterInfo 集群信息对象

集群性能计量

调用方法 checkClusterMetrics() 可以返回集群详细性能指标信息：

guard let m = try yarn.checkClusterMetrics() else {
    // something wrong
}
print(m.availableMB)
print(m.availableVirtualCores)
print(m.allocatedVirtualCores)
print(m.totalMB)

一旦调用成功， checkClusterMetrics() 会返回如下的集群性能指标对象：

ClusterMetrics 集群性能指标对象

集群调度信息

函数 checkSchedulerInfo() 用于返回集群调度详细信息

guard let sch = try yarn.checkSchedulerInfo() else {
    // 出错了
}
print(sch.capacity)
print(sch.maxCapacity)
print(sch.queueName)
print(sch.queues.count)

如果调用成功， checkSchedulerInfo() 会返回一个 SchedulerInfo 数据结构，如下所示：

SchedulerInfo 调度信息对象

FairQueue 根队列对象

队列对象

用户对象

ResourcesUsed 已占用资源

集群节点

查看所有节点

调用方法 checkClusterNodes() 能够以数组形式返回一个集群中的所有节点信息。

let nodes = try yarn.checkClusterNodes()
nodes.forEach { node in
    print(node.rack)
    print(node.availableVirtualCores)
    print(node.availMemoryMB)
    print(node.healthReport)
    print(node.healthStatus)
    print(node.id)
    print(node.lastHealthUpdate)
    print(node.nodeHostName)
    print(node.nodeHTTPAddress)

调用完成后，checkClusterNodes() 返回的节点对象数据结构如以下章节所述：

节点对象

Check A Node

调用方法 checkClusterNode()可获得具体节点的详细数据结构。

guard let n = try yarn.checkClusterNode(id: "host.domain.com:8041") else {
    // 出错了
}

集群应用程序

查看所有应用程序

调用方法 checkApps() 可以获得集群上所有应用程序：

let apps = try yarn.checkApps()
// 或者可以增加下列查询参数以缩小查询范围：
/// let apps = try yarn.checkApps(states: [APP.State.FINISHED, APP.State.RUNNING], finalStatus: APP.FinalStatus.SUCCEEDED)

apps.forEach{ a in
    print(a.allocatedMB)
    print(a.allocatedVCores)
    print(a.amContainerLogs)
    print(a.amHostHttpAddress)
    print(a.amNodeLabelExpression)
    print(a.amRPCAddress)
    print(a.applicationPriority)
    print(a.applicationTags)
}

一旦成功调用， checkApps() 方法会返回一个应用程序对象组成的数组，每个对象的数据结构说明如下：

应用程序对象

查看特定应用程序

调用方法 checkApp() 可以用于查看特定应用程序：

let a = try yarn.checkApp(id: "application_1484231633049_0025")
print(a.allocatedMB)
print(a.allocatedVCores)
print(a.amContainerLogs)
print(a.amHostHttpAddress)
print(a.amNodeLabelExpression)
print(a.amRPCAddress)
print(a.applicationPriority)
print(a.applicationTags)

返回值为应用程序结构，如上文所述。

申请一个新应用程序

调用方法 newApplication() 可以返回一个新建的应用程序句柄。

guard let a = try yarn.newApplication() else {
    // 新建应用程序失败
}

一旦调用成功，该方法 newApplication() 会返回一个 NewApplication 新程序对象，如下所述：

NewApplication 新程序对象

ResourcesUsed 资源用量数据结构

新建的 NewApplication 对象会包含一个名为 ResourceUsed 的特殊对象，如下所述：

修改新建程序模板并正式提交为应用程序

在新建应用程序基础上，修改该程序的具体内容并调用 submit() 方法能够将修改结果提交为一个可以真正运行的应用程序。

⚠️注意⚠️ Yarn映射归并应用程序提交的详细方法已经超过了本文的范围，详细请参考下一代 Hadoop 映射归并程序——自定义YARN应用 https://wiki.apache.org/hadoop/WritingYarnApps

// 首先创建一个空应用程序模板
let sum = SubmitApplication()

// *必须* 设置应用程序编号
sum.id =  "application_1484231633049_0025"
// 设置应用程序名称
sum.name = "test"

// 分配空白的本地资源
let local = LocalResource(resource: "hdfs://localhost:9000/user/rockywei/DistributedShell/demo-app/AppMaster.jar", type: .FILE, visibility: .APPLICATION, size: 43004, timestamp: 1405452071209)

// 将资源信息指定到一个数组
let localResources = Entries([Entry(key:"AppMaster.jar", value: local)])

// *必须* 填写应用程序的映射命令
let commands = Commands("/hdp/bin/hadoop jar /hdp/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.0.0-alpha1.jar grep input output 'dfs[a-z.]+'")

// 设置环境变量
let environments = Entries([Entry(key:"DISTRIBUTEDSHELLSCRIPTTIMESTAMP", value: "1405459400754"), Entry(key:"CLASSPATH", value:"{{CLASSPATH}}<CPS>./*<CPS>{{HADOOP_CONF_DIR}}<CPS>{{HADOOP_COMMON_HOME}}/share/hadoop/common/*<CPS>{{HADOOP_COMMON_HOME}}/share/hadoop/common/lib/*<CPS>{{HADOOP_HDFS_HOME}}/share/hadoop/hdfs/*<CPS>{{HADOOP_HDFS_HOME}}/share/hadoop/hdfs/lib/*<CPS>{{HADOOP_YARN_HOME}}/share/hadoop/yarn/*<CPS>{{HADOOP_YARN_HOME}}/share/hadoop/yarn/lib/*<CPS>./log4j.properties"), Entry(key:"DISTRIBUTEDSHELLSCRIPTLEN", value:6), Entry(key:"DISTRIBUTEDSHELLSCRIPTLOCATION", value: "hdfs://localhost:9000/user/rockywei/demo-app/shellCommands")])

// 设置容器信息
sum.amContainerSpec = AmContainerSpec(localResources: localResources, environment: environments, commands: commands)

// 设置其他信息
sum.unmanagedAM = false
sum.maxAppAttempts = 2
sum.resource = ResourceRequest(memory: 1024, vCores: 1)
sum.type = "MapReduce"
sum.keepContainersAcrossApplicationAttempts = false

// 设置日志环境
sum.logAggregationContext = LogAggregationContext(logIncludePattern: "file1", logExcludePattern: "file2", rolledLogIncludePattern: "file3", rolledLogExcludePattern: "file4", logAggregationPolicyClassName: "org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation.AllContainerLogAggregationPolicy", logAggregationPolicyParameters: "")

// 设置验证失败间隔（1小时）
sum.attemptFailuresValidityInterval = 3600000

// 如果可能的话，设置资源预留代码
sum.reservationId = "reservation_1454114874_1"
sum.amBlackListingRequests = AmBlackListingRequests(amBlackListingEnabled: true, disableFailureThreshold: 0.01)

try yarn.submit(application: sum)

SubmitApplication 待提交应用程序对象

SubmitApplication 类包括了一系列子类以完成应用程序的内容组成，详见下文：

AmContainerSpec 应用程序容器指标对象

LocalResource 本地资源对象

Commands 命令对象

目前 Commands 命令对象只有一个名为 command 的字段。 ⚠️NOTE⚠️ 根据 Hadoop 3.0 alpha 参考手册，commands 应该为一组命令行组成的字符串数组，但是该参考手册中给出的范例却只有一个有效的字符串作为命令行，因此该功能为试验性，有可能根据情况随时升级。

Credentials 密码对象

ResourceRequest 资源请求对象

Entries 字典对象

Entries 对象只有一个元素entry，可以是一个Entry普通字典或者EncryptedEntry加密字典。 每个 Entry 有一对表达式组成： key是条目名称，value是对应的值，类型是Any任意类型。

而两种不同字典 Entry 普通字典和 EncryptedEntry 加密字典的区别是EncryptedEntry 会在提交到应用程序前进行base64加密，只要值的类型是 String 字符串或者[UInt8]二进制数组。

AmBlackListingRequests 应用程序主服务器黑名单请求

LogAggregationContext 日志集合环境对象

应用程序状态控制

调用方法 getApplicationStatus() 会返回当前应用程序的状态。

guard let state = try yarn.getApplicationStatus(id:  "application_1484231633049_0025") else {
    // 出错了
}
print(state)

方法 setApplicationStatus() 用于将应用程序的状态调整为预期设置。

try yarn.setApplicationStatus(id:  "application_1484231633049_0025", state: .KILLED)

有效的状态包括：NEW（新建）、NEW_SAVING（正在保存新建应用）、SUBMITTED（已经提交）、ACCEPTED（已经接受）、RUNNING（正在运行）、FINISHED（运行结束）、FAILED（运行失败）、KILLED（应用程序已经被终止/关闭）

应用程序队列控制

调用方法 getApplicationQueue() 可返回当前应用程序队列名称：

guard let queue = try yarn.getApplicationQueue(id:  "application_1484231633049_0025") else {
    // 出错了
}
print(queue)

方法 setApplicationQueue() 则可用于设置当前队列的名称。

try yarn.setApplicationQueue(id:  "application_1484231633049_0025", queue:"a1a")

应用程序优先级控制

调用方法 getApplicationPriority() 可以返回当前应用程序的优先级

guard let priority = try yarn.getApplicationPriority(id:  "application_1484231633049_0025") else {
    // 出错了
}
print(priority)

当前优先级返回结果是一个整数，比如零。

调用方法 setApplicationPriority()能用于设置当前应用程序的优先级：

try yarn.setApplicationPriority(id:  "application_1484231633049_0025", priority: 1)

应用程序尝试

调用方法 checkAppAttempts() 可以返回一组应用程序的尝试信息：

guard let attempts = try yarn.checkAppAttempts(id:  "application_1484231633049_0025") else {
    // 出错了
}
attempts.forEach { attempt in
    print(attempt.containerId)
    print(attempt.id)
    print(attempt.nodeHttpAddress)
    print(attempt.nodeId)
    print(attempt.startTime)
}//next

一旦成功调用 checkAppAttempts() 会返回 AppAttempt 应用程序尝试对象数组，其数据结构如下所示：

AppAttempt 应用程序尝试对象

应用程序统计数据

调用方法 checkAppStatistics() 可以返回一组应用程序的统计指标变量。

let sta = try yarn.checkAppStatistics(states: [APP.State.FINISHED, APP.State.RUNNING])
sta.forEach{ s in
    print(s.count)
    print(s.state)
    print(s.type)
}//next s

checkAppStatistics() 函数允许两个额外参数：