001/** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.mapreduce.tools; 019 020import java.io.BufferedOutputStream; 021import java.io.File; 022import java.io.FileOutputStream; 023import java.io.IOException; 024import java.io.OutputStreamWriter; 025import java.io.PrintStream; 026import java.io.PrintWriter; 027import java.util.ArrayList; 028import java.util.Arrays; 029import java.util.HashSet; 030import java.util.List; 031import java.util.Set; 032 033import com.google.common.annotations.VisibleForTesting; 034import org.apache.commons.lang.StringUtils; 035import org.apache.commons.logging.Log; 036import org.apache.commons.logging.LogFactory; 037import org.apache.hadoop.classification.InterfaceAudience; 038import org.apache.hadoop.classification.InterfaceStability; 039import org.apache.hadoop.classification.InterfaceAudience.Private; 040import org.apache.hadoop.conf.Configuration; 041import org.apache.hadoop.conf.Configured; 042import org.apache.hadoop.ipc.RemoteException; 043import org.apache.hadoop.mapred.JobConf; 044import org.apache.hadoop.mapred.TIPStatus; 045import org.apache.hadoop.mapreduce.Cluster; 046import org.apache.hadoop.mapreduce.Counters; 047import org.apache.hadoop.mapreduce.Job; 048import org.apache.hadoop.mapreduce.JobID; 049import org.apache.hadoop.mapreduce.JobPriority; 050import org.apache.hadoop.mapreduce.JobStatus; 051import org.apache.hadoop.mapreduce.MRJobConfig; 052import org.apache.hadoop.mapreduce.TaskAttemptID; 053import org.apache.hadoop.mapreduce.TaskCompletionEvent; 054import org.apache.hadoop.mapreduce.TaskReport; 055import org.apache.hadoop.mapreduce.TaskTrackerInfo; 056import org.apache.hadoop.mapreduce.TaskType; 057import org.apache.hadoop.mapreduce.jobhistory.HistoryViewer; 058import org.apache.hadoop.mapreduce.v2.LogParams; 059import org.apache.hadoop.security.AccessControlException; 060import org.apache.hadoop.util.ExitUtil; 061import org.apache.hadoop.util.Tool; 062import org.apache.hadoop.util.ToolRunner; 063import org.apache.hadoop.yarn.logaggregation.LogCLIHelpers; 064 065import com.google.common.base.Charsets; 066 067/** 068 * Interprets the map reduce cli options 069 */ 070@InterfaceAudience.Public 071@InterfaceStability.Stable 072public class CLI extends Configured implements Tool { 073 private static final Log LOG = LogFactory.getLog(CLI.class); 074 protected Cluster cluster; 075 private static final Set<String> taskTypes = new HashSet<String>( 076 Arrays.asList("MAP", "REDUCE")); 077 private final Set<String> taskStates = new HashSet<String>(Arrays.asList( 078 "running", "completed", "pending", "failed", "killed")); 079 080 public CLI() { 081 } 082 083 public CLI(Configuration conf) { 084 setConf(conf); 085 } 086 087 public int run(String[] argv) throws Exception { 088 int exitCode = -1; 089 if (argv.length < 1) { 090 displayUsage(""); 091 return exitCode; 092 } 093 // process arguments 094 String cmd = argv[0]; 095 String submitJobFile = null; 096 String jobid = null; 097 String taskid = null; 098 String historyFileOrJobId = null; 099 String historyOutFile = null; 100 String historyOutFormat = HistoryViewer.HUMAN_FORMAT; 101 String counterGroupName = null; 102 String counterName = null; 103 JobPriority jp = null; 104 String taskType = null; 105 String taskState = null; 106 int fromEvent = 0; 107 int nEvents = 0; 108 boolean getStatus = false; 109 boolean getCounter = false; 110 boolean killJob = false; 111 boolean listEvents = false; 112 boolean viewHistory = false; 113 boolean viewAllHistory = false; 114 boolean listJobs = false; 115 boolean listAllJobs = false; 116 boolean listActiveTrackers = false; 117 boolean listBlacklistedTrackers = false; 118 boolean displayTasks = false; 119 boolean killTask = false; 120 boolean failTask = false; 121 boolean setJobPriority = false; 122 boolean logs = false; 123 124 if ("-submit".equals(cmd)) { 125 if (argv.length != 2) { 126 displayUsage(cmd); 127 return exitCode; 128 } 129 submitJobFile = argv[1]; 130 } else if ("-status".equals(cmd)) { 131 if (argv.length != 2) { 132 displayUsage(cmd); 133 return exitCode; 134 } 135 jobid = argv[1]; 136 getStatus = true; 137 } else if("-counter".equals(cmd)) { 138 if (argv.length != 4) { 139 displayUsage(cmd); 140 return exitCode; 141 } 142 getCounter = true; 143 jobid = argv[1]; 144 counterGroupName = argv[2]; 145 counterName = argv[3]; 146 } else if ("-kill".equals(cmd)) { 147 if (argv.length != 2) { 148 displayUsage(cmd); 149 return exitCode; 150 } 151 jobid = argv[1]; 152 killJob = true; 153 } else if ("-set-priority".equals(cmd)) { 154 if (argv.length != 3) { 155 displayUsage(cmd); 156 return exitCode; 157 } 158 jobid = argv[1]; 159 try { 160 jp = JobPriority.valueOf(argv[2]); 161 } catch (IllegalArgumentException iae) { 162 LOG.info(iae); 163 displayUsage(cmd); 164 return exitCode; 165 } 166 setJobPriority = true; 167 } else if ("-events".equals(cmd)) { 168 if (argv.length != 4) { 169 displayUsage(cmd); 170 return exitCode; 171 } 172 jobid = argv[1]; 173 fromEvent = Integer.parseInt(argv[2]); 174 nEvents = Integer.parseInt(argv[3]); 175 listEvents = true; 176 } else if ("-history".equals(cmd)) { 177 viewHistory = true; 178 if (argv.length < 2 || argv.length > 7) { 179 displayUsage(cmd); 180 return exitCode; 181 } 182 183 // Some arguments are optional while others are not, and some require 184 // second arguments. Due to this, the indexing can vary depending on 185 // what's specified and what's left out, as summarized in the below table: 186 // [all] <jobHistoryFile|jobId> [-outfile <file>] [-format <human|json>] 187 // 1 2 3 4 5 6 188 // 1 2 3 4 189 // 1 2 3 4 190 // 1 2 191 // 1 2 3 4 5 192 // 1 2 3 193 // 1 2 3 194 // 1 195 196 // "all" is optional, but comes first if specified 197 int index = 1; 198 if ("all".equals(argv[index])) { 199 index++; 200 viewAllHistory = true; 201 if (argv.length == 2) { 202 displayUsage(cmd); 203 return exitCode; 204 } 205 } 206 // Get the job history file or job id argument 207 historyFileOrJobId = argv[index++]; 208 // "-outfile" is optional, but if specified requires a second argument 209 if (argv.length > index + 1 && "-outfile".equals(argv[index])) { 210 index++; 211 historyOutFile = argv[index++]; 212 } 213 // "-format" is optional, but if specified required a second argument 214 if (argv.length > index + 1 && "-format".equals(argv[index])) { 215 index++; 216 historyOutFormat = argv[index++]; 217 } 218 // Check for any extra arguments that don't belong here 219 if (argv.length > index) { 220 displayUsage(cmd); 221 return exitCode; 222 } 223 } else if ("-list".equals(cmd)) { 224 if (argv.length != 1 && !(argv.length == 2 && "all".equals(argv[1]))) { 225 displayUsage(cmd); 226 return exitCode; 227 } 228 if (argv.length == 2 && "all".equals(argv[1])) { 229 listAllJobs = true; 230 } else { 231 listJobs = true; 232 } 233 } else if("-kill-task".equals(cmd)) { 234 if (argv.length != 2) { 235 displayUsage(cmd); 236 return exitCode; 237 } 238 killTask = true; 239 taskid = argv[1]; 240 } else if("-fail-task".equals(cmd)) { 241 if (argv.length != 2) { 242 displayUsage(cmd); 243 return exitCode; 244 } 245 failTask = true; 246 taskid = argv[1]; 247 } else if ("-list-active-trackers".equals(cmd)) { 248 if (argv.length != 1) { 249 displayUsage(cmd); 250 return exitCode; 251 } 252 listActiveTrackers = true; 253 } else if ("-list-blacklisted-trackers".equals(cmd)) { 254 if (argv.length != 1) { 255 displayUsage(cmd); 256 return exitCode; 257 } 258 listBlacklistedTrackers = true; 259 } else if ("-list-attempt-ids".equals(cmd)) { 260 if (argv.length != 4) { 261 displayUsage(cmd); 262 return exitCode; 263 } 264 jobid = argv[1]; 265 taskType = argv[2]; 266 taskState = argv[3]; 267 displayTasks = true; 268 if (!taskTypes.contains(taskType.toUpperCase())) { 269 System.out.println("Error: Invalid task-type: " + taskType); 270 displayUsage(cmd); 271 return exitCode; 272 } 273 if (!taskStates.contains(taskState.toLowerCase())) { 274 System.out.println("Error: Invalid task-state: " + taskState); 275 displayUsage(cmd); 276 return exitCode; 277 } 278 } else if ("-logs".equals(cmd)) { 279 if (argv.length == 2 || argv.length ==3) { 280 logs = true; 281 jobid = argv[1]; 282 if (argv.length == 3) { 283 taskid = argv[2]; 284 } else { 285 taskid = null; 286 } 287 } else { 288 displayUsage(cmd); 289 return exitCode; 290 } 291 } else { 292 displayUsage(cmd); 293 return exitCode; 294 } 295 296 // initialize cluster 297 cluster = createCluster(); 298 299 // Submit the request 300 try { 301 if (submitJobFile != null) { 302 Job job = Job.getInstance(new JobConf(submitJobFile)); 303 job.submit(); 304 System.out.println("Created job " + job.getJobID()); 305 exitCode = 0; 306 } else if (getStatus) { 307 Job job = getJob(JobID.forName(jobid)); 308 if (job == null) { 309 System.out.println("Could not find job " + jobid); 310 } else { 311 Counters counters = job.getCounters(); 312 System.out.println(); 313 System.out.println(job); 314 if (counters != null) { 315 System.out.println(counters); 316 } else { 317 System.out.println("Counters not available. Job is retired."); 318 } 319 exitCode = 0; 320 } 321 } else if (getCounter) { 322 Job job = getJob(JobID.forName(jobid)); 323 if (job == null) { 324 System.out.println("Could not find job " + jobid); 325 } else { 326 Counters counters = job.getCounters(); 327 if (counters == null) { 328 System.out.println("Counters not available for retired job " + 329 jobid); 330 exitCode = -1; 331 } else { 332 System.out.println(getCounter(counters, 333 counterGroupName, counterName)); 334 exitCode = 0; 335 } 336 } 337 } else if (killJob) { 338 Job job = getJob(JobID.forName(jobid)); 339 if (job == null) { 340 System.out.println("Could not find job " + jobid); 341 } else { 342 job.killJob(); 343 System.out.println("Killed job " + jobid); 344 exitCode = 0; 345 } 346 } else if (setJobPriority) { 347 Job job = getJob(JobID.forName(jobid)); 348 if (job == null) { 349 System.out.println("Could not find job " + jobid); 350 } else { 351 job.setPriority(jp); 352 System.out.println("Changed job priority."); 353 exitCode = 0; 354 } 355 } else if (viewHistory) { 356 // If it ends with .jhist, assume it's a jhist file; otherwise, assume 357 // it's a Job ID 358 if (historyFileOrJobId.endsWith(".jhist")) { 359 viewHistory(historyFileOrJobId, viewAllHistory, historyOutFile, 360 historyOutFormat); 361 exitCode = 0; 362 } else { 363 Job job = getJob(JobID.forName(historyFileOrJobId)); 364 if (job == null) { 365 System.out.println("Could not find job " + jobid); 366 } else { 367 String historyUrl = job.getHistoryUrl(); 368 if (historyUrl == null || historyUrl.isEmpty()) { 369 System.out.println("History file for job " + historyFileOrJobId + 370 " is currently unavailable."); 371 } else { 372 viewHistory(historyUrl, viewAllHistory, historyOutFile, 373 historyOutFormat); 374 exitCode = 0; 375 } 376 } 377 } 378 } else if (listEvents) { 379 listEvents(getJob(JobID.forName(jobid)), fromEvent, nEvents); 380 exitCode = 0; 381 } else if (listJobs) { 382 listJobs(cluster); 383 exitCode = 0; 384 } else if (listAllJobs) { 385 listAllJobs(cluster); 386 exitCode = 0; 387 } else if (listActiveTrackers) { 388 listActiveTrackers(cluster); 389 exitCode = 0; 390 } else if (listBlacklistedTrackers) { 391 listBlacklistedTrackers(cluster); 392 exitCode = 0; 393 } else if (displayTasks) { 394 displayTasks(getJob(JobID.forName(jobid)), taskType, taskState); 395 exitCode = 0; 396 } else if(killTask) { 397 TaskAttemptID taskID = TaskAttemptID.forName(taskid); 398 Job job = getJob(taskID.getJobID()); 399 if (job == null) { 400 System.out.println("Could not find job " + jobid); 401 } else if (job.killTask(taskID, false)) { 402 System.out.println("Killed task " + taskid); 403 exitCode = 0; 404 } else { 405 System.out.println("Could not kill task " + taskid); 406 exitCode = -1; 407 } 408 } else if(failTask) { 409 TaskAttemptID taskID = TaskAttemptID.forName(taskid); 410 Job job = getJob(taskID.getJobID()); 411 if (job == null) { 412 System.out.println("Could not find job " + jobid); 413 } else if(job.killTask(taskID, true)) { 414 System.out.println("Killed task " + taskID + " by failing it"); 415 exitCode = 0; 416 } else { 417 System.out.println("Could not fail task " + taskid); 418 exitCode = -1; 419 } 420 } else if (logs) { 421 try { 422 JobID jobID = JobID.forName(jobid); 423 TaskAttemptID taskAttemptID = TaskAttemptID.forName(taskid); 424 LogParams logParams = cluster.getLogParams(jobID, taskAttemptID); 425 LogCLIHelpers logDumper = new LogCLIHelpers(); 426 logDumper.setConf(getConf()); 427 exitCode = logDumper.dumpAContainersLogs(logParams.getApplicationId(), 428 logParams.getContainerId(), logParams.getNodeId(), 429 logParams.getOwner()); 430 } catch (IOException e) { 431 if (e instanceof RemoteException) { 432 throw e; 433 } 434 System.out.println(e.getMessage()); 435 } 436 } 437 } catch (RemoteException re) { 438 IOException unwrappedException = re.unwrapRemoteException(); 439 if (unwrappedException instanceof AccessControlException) { 440 System.out.println(unwrappedException.getMessage()); 441 } else { 442 throw re; 443 } 444 } finally { 445 cluster.close(); 446 } 447 return exitCode; 448 } 449 450 Cluster createCluster() throws IOException { 451 return new Cluster(getConf()); 452 } 453 454 private String getJobPriorityNames() { 455 StringBuffer sb = new StringBuffer(); 456 for (JobPriority p : JobPriority.values()) { 457 sb.append(p.name()).append(" "); 458 } 459 return sb.substring(0, sb.length()-1); 460 } 461 462 private String getTaskTypes() { 463 return StringUtils.join(taskTypes, " "); 464 } 465 466 /** 467 * Display usage of the command-line tool and terminate execution. 468 */ 469 private void displayUsage(String cmd) { 470 String prefix = "Usage: CLI "; 471 String jobPriorityValues = getJobPriorityNames(); 472 String taskStates = "running, completed"; 473 474 if ("-submit".equals(cmd)) { 475 System.err.println(prefix + "[" + cmd + " <job-file>]"); 476 } else if ("-status".equals(cmd) || "-kill".equals(cmd)) { 477 System.err.println(prefix + "[" + cmd + " <job-id>]"); 478 } else if ("-counter".equals(cmd)) { 479 System.err.println(prefix + "[" + cmd + 480 " <job-id> <group-name> <counter-name>]"); 481 } else if ("-events".equals(cmd)) { 482 System.err.println(prefix + "[" + cmd + 483 " <job-id> <from-event-#> <#-of-events>]. Event #s start from 1."); 484 } else if ("-history".equals(cmd)) { 485 System.err.println(prefix + "[" + cmd + " [all] <jobHistoryFile|jobId> " + 486 "[-outfile <file>] [-format <human|json>]]"); 487 } else if ("-list".equals(cmd)) { 488 System.err.println(prefix + "[" + cmd + " [all]]"); 489 } else if ("-kill-task".equals(cmd) || "-fail-task".equals(cmd)) { 490 System.err.println(prefix + "[" + cmd + " <task-attempt-id>]"); 491 } else if ("-set-priority".equals(cmd)) { 492 System.err.println(prefix + "[" + cmd + " <job-id> <priority>]. " + 493 "Valid values for priorities are: " 494 + jobPriorityValues); 495 } else if ("-list-active-trackers".equals(cmd)) { 496 System.err.println(prefix + "[" + cmd + "]"); 497 } else if ("-list-blacklisted-trackers".equals(cmd)) { 498 System.err.println(prefix + "[" + cmd + "]"); 499 } else if ("-list-attempt-ids".equals(cmd)) { 500 System.err.println(prefix + "[" + cmd + 501 " <job-id> <task-type> <task-state>]. " + 502 "Valid values for <task-type> are " + getTaskTypes() + ". " + 503 "Valid values for <task-state> are " + taskStates); 504 } else if ("-logs".equals(cmd)) { 505 System.err.println(prefix + "[" + cmd + 506 " <job-id> <task-attempt-id>]. " + 507 " <task-attempt-id> is optional to get task attempt logs."); 508 } else { 509 System.err.printf(prefix + "<command> <args>%n"); 510 System.err.printf("\t[-submit <job-file>]%n"); 511 System.err.printf("\t[-status <job-id>]%n"); 512 System.err.printf("\t[-counter <job-id> <group-name> <counter-name>]%n"); 513 System.err.printf("\t[-kill <job-id>]%n"); 514 System.err.printf("\t[-set-priority <job-id> <priority>]. " + 515 "Valid values for priorities are: " + jobPriorityValues + "%n"); 516 System.err.printf("\t[-events <job-id> <from-event-#> <#-of-events>]%n"); 517 System.err.printf("\t[-history [all] <jobHistoryFile|jobId> " + 518 "[-outfile <file>] [-format <human|json>]]%n"); 519 System.err.printf("\t[-list [all]]%n"); 520 System.err.printf("\t[-list-active-trackers]%n"); 521 System.err.printf("\t[-list-blacklisted-trackers]%n"); 522 System.err.println("\t[-list-attempt-ids <job-id> <task-type> " + 523 "<task-state>]. " + 524 "Valid values for <task-type> are " + getTaskTypes() + ". " + 525 "Valid values for <task-state> are " + taskStates); 526 System.err.printf("\t[-kill-task <task-attempt-id>]%n"); 527 System.err.printf("\t[-fail-task <task-attempt-id>]%n"); 528 System.err.printf("\t[-logs <job-id> <task-attempt-id>]%n%n"); 529 ToolRunner.printGenericCommandUsage(System.out); 530 } 531 } 532 533 private void viewHistory(String historyFile, boolean all, 534 String historyOutFile, String format) throws IOException { 535 HistoryViewer historyViewer = new HistoryViewer(historyFile, 536 getConf(), all, format); 537 PrintStream ps = System.out; 538 if (historyOutFile != null) { 539 ps = new PrintStream(new BufferedOutputStream(new FileOutputStream( 540 new File(historyOutFile))), true, "UTF-8"); 541 } 542 historyViewer.print(ps); 543 } 544 545 protected long getCounter(Counters counters, String counterGroupName, 546 String counterName) throws IOException { 547 return counters.findCounter(counterGroupName, counterName).getValue(); 548 } 549 550 /** 551 * List the events for the given job 552 * @param jobId the job id for the job's events to list 553 * @throws IOException 554 */ 555 private void listEvents(Job job, int fromEventId, int numEvents) 556 throws IOException, InterruptedException { 557 TaskCompletionEvent[] events = job. 558 getTaskCompletionEvents(fromEventId, numEvents); 559 System.out.println("Task completion events for " + job.getJobID()); 560 System.out.println("Number of events (from " + fromEventId + ") are: " 561 + events.length); 562 for(TaskCompletionEvent event: events) { 563 System.out.println(event.getStatus() + " " + 564 event.getTaskAttemptId() + " " + 565 getTaskLogURL(event.getTaskAttemptId(), event.getTaskTrackerHttp())); 566 } 567 } 568 569 protected static String getTaskLogURL(TaskAttemptID taskId, String baseUrl) { 570 return (baseUrl + "/tasklog?plaintext=true&attemptid=" + taskId); 571 } 572 573 @VisibleForTesting 574 Job getJob(JobID jobid) throws IOException, InterruptedException { 575 576 int maxRetry = getConf().getInt(MRJobConfig.MR_CLIENT_JOB_MAX_RETRIES, 577 MRJobConfig.DEFAULT_MR_CLIENT_JOB_MAX_RETRIES); 578 long retryInterval = getConf() 579 .getLong(MRJobConfig.MR_CLIENT_JOB_RETRY_INTERVAL, 580 MRJobConfig.DEFAULT_MR_CLIENT_JOB_RETRY_INTERVAL); 581 Job job = cluster.getJob(jobid); 582 583 for (int i = 0; i < maxRetry; ++i) { 584 if (job != null) { 585 return job; 586 } 587 LOG.info("Could not obtain job info after " + String.valueOf(i + 1) 588 + " attempt(s). Sleeping for " + String.valueOf(retryInterval / 1000) 589 + " seconds and retrying."); 590 Thread.sleep(retryInterval); 591 job = cluster.getJob(jobid); 592 } 593 return job; 594 } 595 596 597 /** 598 * Dump a list of currently running jobs 599 * @throws IOException 600 */ 601 private void listJobs(Cluster cluster) 602 throws IOException, InterruptedException { 603 List<JobStatus> runningJobs = new ArrayList<JobStatus>(); 604 for (JobStatus job : cluster.getAllJobStatuses()) { 605 if (!job.isJobComplete()) { 606 runningJobs.add(job); 607 } 608 } 609 displayJobList(runningJobs.toArray(new JobStatus[0])); 610 } 611 612 /** 613 * Dump a list of all jobs submitted. 614 * @throws IOException 615 */ 616 private void listAllJobs(Cluster cluster) 617 throws IOException, InterruptedException { 618 displayJobList(cluster.getAllJobStatuses()); 619 } 620 621 /** 622 * Display the list of active trackers 623 */ 624 private void listActiveTrackers(Cluster cluster) 625 throws IOException, InterruptedException { 626 TaskTrackerInfo[] trackers = cluster.getActiveTaskTrackers(); 627 for (TaskTrackerInfo tracker : trackers) { 628 System.out.println(tracker.getTaskTrackerName()); 629 } 630 } 631 632 /** 633 * Display the list of blacklisted trackers 634 */ 635 private void listBlacklistedTrackers(Cluster cluster) 636 throws IOException, InterruptedException { 637 TaskTrackerInfo[] trackers = cluster.getBlackListedTaskTrackers(); 638 if (trackers.length > 0) { 639 System.out.println("BlackListedNode \t Reason"); 640 } 641 for (TaskTrackerInfo tracker : trackers) { 642 System.out.println(tracker.getTaskTrackerName() + "\t" + 643 tracker.getReasonForBlacklist()); 644 } 645 } 646 647 private void printTaskAttempts(TaskReport report) { 648 if (report.getCurrentStatus() == TIPStatus.COMPLETE) { 649 System.out.println(report.getSuccessfulTaskAttemptId()); 650 } else if (report.getCurrentStatus() == TIPStatus.RUNNING) { 651 for (TaskAttemptID t : 652 report.getRunningTaskAttemptIds()) { 653 System.out.println(t); 654 } 655 } 656 } 657 658 /** 659 * Display the information about a job's tasks, of a particular type and 660 * in a particular state 661 * 662 * @param job the job 663 * @param type the type of the task (map/reduce/setup/cleanup) 664 * @param state the state of the task 665 * (pending/running/completed/failed/killed) 666 */ 667 protected void displayTasks(Job job, String type, String state) 668 throws IOException, InterruptedException { 669 TaskReport[] reports = job.getTaskReports(TaskType.valueOf(type.toUpperCase())); 670 for (TaskReport report : reports) { 671 TIPStatus status = report.getCurrentStatus(); 672 if ((state.equalsIgnoreCase("pending") && status ==TIPStatus.PENDING) || 673 (state.equalsIgnoreCase("running") && status ==TIPStatus.RUNNING) || 674 (state.equalsIgnoreCase("completed") && status == TIPStatus.COMPLETE) || 675 (state.equalsIgnoreCase("failed") && status == TIPStatus.FAILED) || 676 (state.equalsIgnoreCase("killed") && status == TIPStatus.KILLED)) { 677 printTaskAttempts(report); 678 } 679 } 680 } 681 682 public void displayJobList(JobStatus[] jobs) 683 throws IOException, InterruptedException { 684 displayJobList(jobs, new PrintWriter(new OutputStreamWriter(System.out, 685 Charsets.UTF_8))); 686 } 687 688 @Private 689 public static String headerPattern = "%23s\t%10s\t%14s\t%12s\t%12s\t%10s\t%15s\t%15s\t%8s\t%8s\t%10s\t%10s\n"; 690 @Private 691 public static String dataPattern = "%23s\t%10s\t%14d\t%12s\t%12s\t%10s\t%15s\t%15s\t%8s\t%8s\t%10s\t%10s\n"; 692 private static String memPattern = "%dM"; 693 private static String UNAVAILABLE = "N/A"; 694 695 @Private 696 public void displayJobList(JobStatus[] jobs, PrintWriter writer) { 697 writer.println("Total jobs:" + jobs.length); 698 writer.printf(headerPattern, "JobId", "State", "StartTime", "UserName", 699 "Queue", "Priority", "UsedContainers", 700 "RsvdContainers", "UsedMem", "RsvdMem", "NeededMem", "AM info"); 701 for (JobStatus job : jobs) { 702 int numUsedSlots = job.getNumUsedSlots(); 703 int numReservedSlots = job.getNumReservedSlots(); 704 int usedMem = job.getUsedMem(); 705 int rsvdMem = job.getReservedMem(); 706 int neededMem = job.getNeededMem(); 707 writer.printf(dataPattern, 708 job.getJobID().toString(), job.getState(), job.getStartTime(), 709 job.getUsername(), job.getQueue(), 710 job.getPriority().name(), 711 numUsedSlots < 0 ? UNAVAILABLE : numUsedSlots, 712 numReservedSlots < 0 ? UNAVAILABLE : numReservedSlots, 713 usedMem < 0 ? UNAVAILABLE : String.format(memPattern, usedMem), 714 rsvdMem < 0 ? UNAVAILABLE : String.format(memPattern, rsvdMem), 715 neededMem < 0 ? UNAVAILABLE : String.format(memPattern, neededMem), 716 job.getSchedulingInfo()); 717 } 718 writer.flush(); 719 } 720 721 public static void main(String[] argv) throws Exception { 722 int res = ToolRunner.run(new CLI(), argv); 723 ExitUtil.terminate(res); 724 } 725}