PerfectHadoop: MapReduce Application 归并映射应用程序

该项目实现了Hadoop 归并映射应用程序 MapReduce Application Master REST API 的Swift类封装:

• MapReduceApplication(): 获取当前服务器上特定归并映射应用程序状态。

注意事项

⚠️本函数库为试验性模块，可能随时会有更新和调整。 ⚠️

连接到Hadoop映射归并应用程序服务器

为了能够连接到当前活动的Hadoop 映射归并应用程序，请首先用有效的 应用程序代码 初始化一个MapReduceApplication() 对象，并且需要包含必要参数：

// 该连接可能只有基本操作权限
let app = MapReduceApplication(applicationId: "application_12345678_xxxxxx", host: "mapReducer.somedomain.com")

或者您还可以：

// 如果有需要，增加用户名，用于执行更多有权限要求的操作
let app = MapReduceApplication(applicationId: "application_12345678_xxxxxx", host: "mapReducer.somedomain.com", user: "your user name")

用户身份认证

如果服务器需要使用 Kerberos 身份验证，请在代码中加入：

// 将连接设置为KRB5认证模式
let app = MapReduceApplication(applicationId: "application_12345678_xxxxxx", host: "mapReducer.somedomain.com", user: "your user name", auth: .krb5)

MapReduceApplication 映射归并应用程序对象属性

查看基本信息

调用 checkInfo()方法可以获得当前映射归并主应用程序的基本运行情况，返回结果是一个 MapReduceApplication.Info 数据结构：

guard let inf = try app.checkInfo() else {
    // 出错了
}
print(inf.startedOn)
print(inf.hadoopVersion)
print(inf.hadoopBuildVersion)
print(inf.hadoopVersionBuiltOn)

MapReduceApplication.Info 数据结构

MapReduce Job 映射归并作业

调用 checkJobs() 能够返回该应用程序的一个作业数组。 作业数组指的是目前该应用程序所执行的映射归并已完成的作业清单。返回的清单目前并未包含所有参数。最简单的形式就是直接调用 checkJobs() 而且不用包含任何参数，意味着无条件返回所有可以查询到的所有作业：

let jobs = try app.checkJobs()
jobs.forEach { j in
  print(j.id)
  print(j.name)
  print(j.queue)
  print(j.state)
}

如果执行成功，checkJobs() 将返回一系列作业对象，如以下数据结构所述：

作业对象 Job 数据结构

访问控制列表对象 ACL 数据结构

检查特定作业（Job）

可以通过作业代码（Job ID）检查特定作业信息：

let job = try app.checkJob(jobId: "job_1484231633049_0005")

详见 作业（Job）数据结构。

作业尝试 JobAttempt

以下命令能够获得关于某个作业的尝试列表信息：

guard let attempts = try app.checkJobAttempts(jobId: "job_1484231633049_0005") else {
    // 出错了
}
attempts.forEach { attempt in
    print(attempt.id)
    print(attempt.containerId)
    print(attempt.nodeHttpAddress)
    print(attempt.nodeId)
    print(attempt.startTime)
}

JobAttempt 数据结构

作业指标统计（JobCounter）

使用作业指标函数库，您可以获得关于该作业的所有指标信息。

guard let js = try app.checkJobCounters(jobId: "job_1484231633049_0005") else {
    // 出错了
}//end guard
js.counterGroup.forEach{ group in
  print(group.counterGroupName)
  group.counters.forEach { counter in
    print(counter.name)
    print(counter.mapCounterValue)
    print(counter.reduceCounterValue)
    print(counter.totalCounterValue)
  }//next counter
}//next group

JobCounter Object

CounterGroup 指标分组对象

Counter 指标对象

查看作业配置

函数 checkJobConfig() 可以用来检查作业配置情况：

gurard let config = try app.checkJobConfig(jobId: "job_1484231633049_0005") else {
    /// 出错了
}
// print the configuration path
print(config.path)
// check properties of configuration file
for p in config.property {
  print(p.name)
  print(p.value)
  print(p.source)
}

JobConfig 作业配置对象

作业配置对象包含了该作业的资源使用信息。

Property 属性对象

作业子任务（Tasks of A Job）

方法 checkJobTasks() 用于获取特定作业的子任务信息：

// 获得特定作业的所有子任务信息
let tasks = try app.checkJobTasks(jobId: "job_1484231633049_0005")

// print properties of each task
for t in tasks {
  print(t.progress)
  print(t.elapsedTime)
  print(t.state)
  print(t.startTime)
  print(t.id)
  print(t.type)
  print(t.successfulAttempt)
  print(t.finishTime)
}//next t

checkJobTasks() 函数参数

调用 checkJobTasks() 总是需要一个作业编码 jobId ，但是也可以指定第二个附加参数，即子任务类型用于进一步过滤查询结果：

• jobId: 作业编码
• taskType: 可选的子任务类型，即 .MAP 映射任务或者 .REDUCE 归并任务

JobTask Object

函数checkJobTasks()用于返回一个作业子任务数组 JobTask 对象，包括下列属性：

查询特定的作业子任务 JobTask

如果已经获取了有效的 jobId 作业编码和 jobTaskId 作业子任务编码，那么可以使用函数 checkJobTask() 检查特定的作业子任务：

guard let task = try app.checkJobTask(jobId: "job_1484231633049_0005", taskId: "task_1326381300833_2_2_m_0") else {
    // 出错了
}
print(task.progress)

该函数 checkJobTask 返回值为一个 JobTask 作业子任务 对象，如上所述。

作业子任务统计指标

方法 checkJobTaskCounters() 用于返回特定作业子任务的统计指标信息，如以下范例所示：

guard let js = try app.checkJobTaskCounters(jobId: "job_1484231633049_0005", taskId: "task_1326381300833_2_2_m_0") else {
    // 出错了
}
// 打印作业子任务指标编码
print(js.id)
// 检查每个统计指标编组
js.taskCounterGroup.forEach{ group in
  print(group.counterGroupName)
 // 打印每个分组中的详细指标信息
  group.counters.forEach { counter in
    print(counter.name)
    print(counter.value)
  }
}

JobTaskCounter 作业子任务指标对象

Task Attempts 作业子任务尝试

函数 checkJobTaskAttempts() 用于检查特定作业子任务的尝试信息。

let jobTaskAttempts = try app.checkJobTaskAttempts(jobId: "job_1484231633049_0005", taskId: "task_1326381300833_2_2_m_0")
for attempt in jobTaskAttempts {
  print(attempt.id)
  print(attempt.diagnostics)
  print(attempt.assignedContainerId)
  print(attempt.rack)
  print(attempt.state)
  print(attempt.progress)
}//next

一旦成功调用 checkJobTaskAttempts() 会返回一个 TaskAttempt 作业子任务尝试对象，如下所述：

TaskAttempt 作业子任务尝试对象

作业子任务尝试对象 TaskAttempt 包括下列属性：

另外，对于 reduce （归并）作业子任务尝试来说，还可以包括下列属性：

查询特定的 TaskAttempt 作业子任务尝试

如果作业子任务尝试编码有效，则可以执行checkJobTaskAttempt()查询该作业子任务的尝试信息，如下所述：

guard let jobTaskAttempts = try app.checkJobTaskAttempt(jobId: "job_1484231633049_0005", taskId: "task_1326381300833_2_2_m_0", "attempt_1326381300833_2_2_m_0_0") else {
    // 出错了
}//end guard
print(attempt.id)
print(attempt.diagnostics)
print(attempt.assignedContainerId)
print(attempt.rack)
print(attempt.state)
print(attempt.progress)

函数checkJobTaskAttempt() 调用后会返回一个“TaskAttempt 作业子任务尝试对象” 如上文所述。

Attempt State Control 作业尝试状态控制

目前Hadoop 映射归并主程序具有一个试验性质的尝试状态控制。

检查尝试

如果需要查看尝试状态，请调用 checkJobTaskAttemptState() 方法：

guard let state = try checkJobTaskAttemptState(jobId: "job_1484231633049_0005", taskId: "task_1326381300833_2_2_m_0", "attempt_1326381300833_2_2_m_0_0") else {
    // 出错了
}//end guard
print(state)

终止尝试

调用 killTaskAttempt() 方法可以终止尝试：

try killTaskAttempt(jobId: "job_1484231633049_0005", taskId: "task_1326381300833_2_2_m_0", "attempt_1326381300833_2_2_m_0_0")

作业子任务尝试指标统计

方法 checkJobTaskAttemptCounters() 可以获得关于特定作业子任务尝试的详细指标统计，如下文所述：

guard let counters = try app.checkJobTaskAttemptCounters(jobId: "job_1484231633049_0005", taskId: "task_1326381300833_2_2_m_0", "attempt_1326381300833_2_2_m_0_0") else {
    // 出错了
}
// 打印指标计数编码
print(counters.id)
// 遍历所有指标分组
for group in counters.taskAttemptCounterGroup {
  // 打印分组名称
  print(group.counterGroupName)
  // 打印该分组下的所有详细指标
  for counter in group.counters {
    print(counter.name)
    print(counter.value)
  }//next counter
}//next group
}

JobTaskAttemptCounters 作业子任务尝试统计指标对象