PerfectHadoop: YARN Resource Manager

This project provides a Swift wrapper of YARN Resource Manager REST API:

• YARNResourceManager(): access to cluster information of YARN, including cluster and its metrics, scheduler, application submit, etc.

Connect to YARN Resource Manager

To connect to your Hadoop YARN Resource Manager by Perfect, initialize a YARNResourceManager() object with sufficient parameters:

// this connection could possibly do some basic operations
let yarn = YARNResourceManager(host: "yarn.somehadoopdomain.com", port: 8088)

or connect to Hadoop YARN Node Manager with a valid user name:

// add user name if need
let yarn = YARNResourceManager(host: "yarn.somehadoopdomain.com", port: 8088, user: "your user name")

Authentication

If using Kerberos to authenticate, please try codes below:

// set auth to kerberos
let yarn = YARNResourceManager(host: "yarn.somehadoopdomain.com", port: 8088, user: "username", auth: .krb5)

Parameters of YARNResourceManager Object

Get General Information

Call checkClusterInfo() to get the general information of a YARN Resource Manager in form of a ClusterInfo structure:

guard let i = try yarn.checkClusterInfo() else {
    print("unable to check cluster info")
    return
}//end guard
print(i.startedOn)
print(i.state)
print(i.hadoopVersion)
print(i.resourceManagerVersion)

Once called, checkClusterInfo() would return a ClusterInfo object, as described below:

ClusterInfo Object

Cluster Metrics

Method checkClusterMetrics() returns a detailed info structure of cluster.

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

Once called, checkClusterMetrics() would return a ClusterMetrics object as listed below:

ClusterMetrics Object

Cluster Scheduler

Method checkSchedulerInfo() returns a detailed info structure of scheduler.

guard let sch = try yarn.checkSchedulerInfo() else {
    // something wrong, must return
}
print(sch.capacity)
print(sch.maxCapacity)
print(sch.queueName)
print(sch.queues.count)

Once done, checkSchedulerInfo() would return a SchedulerInfo structure as listed below:

SchedulerInfo Object

FairQueue Object

Queue Object

User Object

ResourcesUsed Object

Cluster Nodes

Check All Nodes

Method checkClusterNodes() returns an array of node info of cluster.

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)

Once done, the checkClusterNodes() would return an array of Node object, as described below:

Node Object

Check A Node

Method checkClusterNode() returns a detailed info structure of a node.

guard let n = try yarn.checkClusterNode(id: "host.domain.com:8041") else {
    // something wrong, must return
}

Applications on Cluster

Check All Applications

Method checkApps() returns an array of APP structure. ``` swift let apps = try yarn.checkApps() // or alternatively, you can filter out those APPs you want by setting query parameters: /// 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) } ```

Once done, the checkApps() would return an array of APP objects, as described below:

APP Object

Check A Specific Application

Method checkApp() returns a specific APP Object.

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)

The return value is an APP structure, please check the above APP object for detail.

Apply A New Application

Method newApplication() returns a new application handler.

guard let a = try yarn.newApplication() else {
    // cannot create a new application, must return
}

Once completed, the newApplication() method will return a NewApplication object, as described below:

NewApplication Object

ResourcesUsed Object

The NewApplication object will contain a special object called ResourceUsed, as described below:

Submit Modification to An Application

Method submit() can submit modifications to a specific application.

⚠️Note⚠️ Detail configuration of Yarn MapReduce Application is out of the range of this document, please get more information at Hadoop MapReduce Next Generation - Writing YARN Applications https://wiki.apache.org/hadoop/WritingYarnApps

// create an empty application to fill in the blanks
let sum = SubmitApplication()

// *MUST* set the application id
sum.id =  "application_1484231633049_0025"
// set the application name
sum.name = "test"

// allocate a blank local resource
let local = LocalResource(resource: "hdfs://localhost:9000/user/rockywei/DistributedShell/demo-app/AppMaster.jar", type: .FILE, visibility: .APPLICATION, size: 43004, timestamp: 1405452071209)

// assign the resource into an array
let localResources = Entries([Entry(key:"AppMaster.jar", value: local)])

// *MUST* fill in the field of map reduce command
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.]+'")

// setup environment as need
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")])

// specify the container info
sum.amContainerSpec = AmContainerSpec(localResources: localResources, environment: environments, commands: commands)

// set other information as need
sum.unmanagedAM = false
sum.maxAppAttempts = 2
sum.resource = ResourceRequest(memory: 1024, vCores: 1)
sum.type = "MapReduce"
sum.keepContainersAcrossApplicationAttempts = false

// set the log context
sum.logAggregationContext = LogAggregationContext(logIncludePattern: "file1", logExcludePattern: "file2", rolledLogIncludePattern: "file3", rolledLogExcludePattern: "file4", logAggregationPolicyClassName: "org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation.AllContainerLogAggregationPolicy", logAggregationPolicyParameters: "")

// set the failures validity interval
sum.attemptFailuresValidityInterval = 3600000

// set the reservationId, if possible
sum.reservationId = "reservation_1454114874_1"
sum.amBlackListingRequests = AmBlackListingRequests(amBlackListingEnabled: true, disableFailureThreshold: 0.01)

try yarn.submit(application: sum)

SubmitApplication Object

SubmitApplication class contains quite a few sub types, as described below:

AmContainerSpec Object

LocalResource Object

Commands Object

Currently Commands object only one string field named command. ⚠️NOTE⚠️ According to Hadoop 3.0 alpha document, commands should be an array of command strings with a sequential meaning, however, the example given in Hadoop 3.0 alpha indicates only one command. This feature is experimental and subject to change in future.

Credentials Object

ResourceRequest Object

Entries Object

The Entries Object has only one field of element: entry, which is an array of Entry object or EncryptedEntry object. Each Entry has two elements: a string key and a Any type value.

The difference between Entry and EncryptedEntry is that the value of EncryptedEntry will be encoded in base64 format before submission, providing the type of value is either String or [UInt8].

AmBlackListingRequests Object

LogAggregationContext Object

Application State Control

Method getApplicationStatus() returns the current state of application.

guard let state = try yarn.getApplicationStatus(id:  "application_1484231633049_0025") else {
    // something wrong, must return
}
print(state)

Method setApplicationStatus() can set the current state of application to a designated one.

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

Valid States includes:

NEW, NEW_SAVING, SUBMITTED, ACCEPTED, RUNNING, FINISHED, FAILED, KILLED

Application Queue Control

Method getApplicationQueue() returns the current queue name of application.

guard let queue = try yarn.getApplicationQueue(id:  "application_1484231633049_0025") else {
    // something wrong, must return
}
print(queue)

Method setApplicationQueue() can set the current queue name of application to a designated one.

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

Application Priority Control

Method getApplicationPriority() returns the current priority of application.

guard let priority = try yarn.getApplicationPriority(id:  "application_1484231633049_0025") else {
    // something wrong, must return
}
print(priority)

Currently the returned priority is an integer such as 0.

Method setApplicationPriority() can set the current priority of application to a designated one.

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

Application Attempts

Method checkAppAttempts() returns an array of Attempt object of application.

guard let attempts = try yarn.checkAppAttempts(id:  "application_1484231633049_0025") else {
    // something wrong, must return
}
attempts.forEach { attempt in
    print(attempt.containerId)
    print(attempt.id)
    print(attempt.nodeHttpAddress)
    print(attempt.nodeId)
    print(attempt.startTime)
}//next

Once done, checkAppAttempts() would return an array of AppAttempt object, as described below:

AppAttempt Object

Application Statistics

Method checkAppStatistics() returns an array of statistic variables of application. swift 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() allows user to perform a query with two additional parameters: