开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > Internet/网络编程 > 网格计算 > 查看源码
JobClient.java
资源名称：hadoop-0.20.0.tar.gz　[点击查看]
上传用户：quxuerui
上传日期：2018-01-08
资源大小：41811k
文件大小：62k
源码类别：

网格计算

开发平台：

Java

JobClient.java：源码内容
  1. /**
  2.  * Licensed to the Apache Software Foundation (ASF) under one
  3.  * or more contributor license agreements.  See the NOTICE file
  4.  * distributed with this work for additional information
  5.  * regarding copyright ownership.  The ASF licenses this file
  6.  * to you under the Apache License, Version 2.0 (the
  7.  * "License"); you may not use this file except in compliance
  8.  * with the License.  You may obtain a copy of the License at
  9.  *
  10.  *     http://www.apache.org/licenses/LICENSE-2.0
  11.  *
  12.  * Unless required by applicable law or agreed to in writing, software
  13.  * distributed under the License is distributed on an "AS IS" BASIS,
  14.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15.  * See the License for the specific language governing permissions and
  16.  * limitations under the License.
  17.  */
  18. package org.apache.hadoop.mapred;
  20. import java.io.BufferedReader;
  21. import java.io.BufferedWriter;
  22. import java.io.DataInput;
  23. import java.io.DataOutput;
  24. import java.io.DataOutputStream;
  25. import java.io.FileNotFoundException;
  26. import java.io.FileOutputStream;
  27. import java.io.IOException;
  28. import java.io.InputStream;
  29. import java.io.InputStreamReader;
  30. import java.io.OutputStream;
  31. import java.io.OutputStreamWriter;
  32. import java.net.InetAddress;
  33. import java.net.InetSocketAddress;
  34. import java.net.URI;
  35. import java.net.URISyntaxException;
  36. import java.net.URL;
  37. import java.net.URLConnection;
  38. import java.net.UnknownHostException;
  39. import java.util.Arrays;
  40. import java.util.Collection;
  41. import java.util.Comparator;
  42. import java.util.List;
  44. import javax.security.auth.login.LoginException;
  46. import org.apache.commons.logging.Log;
  47. import org.apache.commons.logging.LogFactory;
  48. import org.apache.hadoop.conf.Configuration;
  49. import org.apache.hadoop.conf.Configured;
  50. import org.apache.hadoop.filecache.DistributedCache;
  51. import org.apache.hadoop.fs.FSDataOutputStream;
  52. import org.apache.hadoop.fs.FileStatus;
  53. import org.apache.hadoop.fs.FileSystem;
  54. import org.apache.hadoop.fs.FileUtil;
  55. import org.apache.hadoop.fs.Path;
  56. import org.apache.hadoop.fs.permission.FsPermission;
  57. import org.apache.hadoop.io.BytesWritable;
  58. import org.apache.hadoop.io.DataOutputBuffer;
  59. import org.apache.hadoop.io.IOUtils;
  60. import org.apache.hadoop.io.Text;
  61. import org.apache.hadoop.io.Writable;
  62. import org.apache.hadoop.io.WritableUtils;
  63. import org.apache.hadoop.io.serializer.SerializationFactory;
  64. import org.apache.hadoop.io.serializer.Serializer;
  65. import org.apache.hadoop.ipc.RPC;
  66. import org.apache.hadoop.mapred.Counters.Counter;
  67. import org.apache.hadoop.mapred.Counters.Group;
  68. import org.apache.hadoop.net.NetUtils;
  69. import org.apache.hadoop.security.UnixUserGroupInformation;
  70. import org.apache.hadoop.util.ReflectionUtils;
  71. import org.apache.hadoop.util.StringUtils;
  72. import org.apache.hadoop.util.Tool;
  73. import org.apache.hadoop.util.ToolRunner;
  75. /**
  76.  * <code>JobClient</code> is the primary interface for the user-job to interact
  77.  * with the {@link JobTracker}.
  78.  * 
  79.  * <code>JobClient</code> provides facilities to submit jobs, track their 
  80.  * progress, access component-tasks' reports/logs, get the Map-Reduce cluster
  81.  * status information etc.
  82.  * 
  83.  * <p>The job submission process involves:
  84.  * <ol>
  85.  *   <li>
  86.  *   Checking the input and output specifications of the job.
  87.  *   </li>
  88.  *   <li>
  89.  *   Computing the {@link InputSplit}s for the job.
  90.  *   </li>
  91.  *   <li>
  92.  *   Setup the requisite accounting information for the {@link DistributedCache} 
  93.  *   of the job, if necessary.
  94.  *   </li>
  95.  *   <li>
  96.  *   Copying the job's jar and configuration to the map-reduce system directory 
  97.  *   on the distributed file-system. 
  98.  *   </li>
  99.  *   <li>
  100.  *   Submitting the job to the <code>JobTracker</code> and optionally monitoring
  101.  *   it's status.
  102.  *   </li>
  103.  * </ol></p>
  104.  *  
  105.  * Normally the user creates the application, describes various facets of the
  106.  * job via {@link JobConf} and then uses the <code>JobClient</code> to submit 
  107.  * the job and monitor its progress.
  108.  * 
  109.  * <p>Here is an example on how to use <code>JobClient</code>:</p>
  110.  * <p><blockquote><pre>
  111.  *     // Create a new JobConf
  112.  *     JobConf job = new JobConf(new Configuration(), MyJob.class);
  113.  *     
  114.  *     // Specify various job-specific parameters     
  115.  *     job.setJobName("myjob");
  116.  *     
  117.  *     job.setInputPath(new Path("in"));
  118.  *     job.setOutputPath(new Path("out"));
  119.  *     
  120.  *     job.setMapperClass(MyJob.MyMapper.class);
  121.  *     job.setReducerClass(MyJob.MyReducer.class);
  122.  *
  123.  *     // Submit the job, then poll for progress until the job is complete
  124.  *     JobClient.runJob(job);
  125.  * </pre></blockquote></p>
  126.  * 
  127.  * <h4 id="JobControl">Job Control</h4>
  128.  * 
  129.  * <p>At times clients would chain map-reduce jobs to accomplish complex tasks 
  130.  * which cannot be done via a single map-reduce job. This is fairly easy since 
  131.  * the output of the job, typically, goes to distributed file-system and that 
  132.  * can be used as the input for the next job.</p>
  133.  * 
  134.  * <p>However, this also means that the onus on ensuring jobs are complete 
  135.  * (success/failure) lies squarely on the clients. In such situations the 
  136.  * various job-control options are:
  137.  * <ol>
  138.  *   <li>
  139.  *   {@link #runJob(JobConf)} : submits the job and returns only after 
  140.  *   the job has completed.
  141.  *   </li>
  142.  *   <li>
  143.  *   {@link #submitJob(JobConf)} : only submits the job, then poll the 
  144.  *   returned handle to the {@link RunningJob} to query status and make 
  145.  *   scheduling decisions.
  146.  *   </li>
  147.  *   <li>
  148.  *   {@link JobConf#setJobEndNotificationURI(String)} : setup a notification
  149.  *   on job-completion, thus avoiding polling.
  150.  *   </li>
  151.  * </ol></p>
  152.  * 
  153.  * @see JobConf
  154.  * @see ClusterStatus
  155.  * @see Tool
  156.  * @see DistributedCache
  157.  */
  158. public class JobClient extends Configured implements MRConstants, Tool  {
  159.   private static final Log LOG = LogFactory.getLog(JobClient.class);
  160.   public static enum TaskStatusFilter { NONE, KILLED, FAILED, SUCCEEDED, ALL }
  161.   private TaskStatusFilter taskOutputFilter = TaskStatusFilter.FAILED; 
  162.   private static final long MAX_JOBPROFILE_AGE = 1000 * 2;
  164.   static{
  165.     Configuration.addDefaultResource("mapred-default.xml");
  166.     Configuration.addDefaultResource("mapred-site.xml");
  167.   }
  169.   /**
  170.    * A NetworkedJob is an implementation of RunningJob.  It holds
  171.    * a JobProfile object to provide some info, and interacts with the
  172.    * remote service to provide certain functionality.
  173.    */
  174.   class NetworkedJob implements RunningJob {
  175.     JobProfile profile;
  176.     JobStatus status;
  177.     long statustime;
  179.     /**
  180.      * We store a JobProfile and a timestamp for when we last
  181.      * acquired the job profile.  If the job is null, then we cannot
  182.      * perform any of the tasks.  The job might be null if the JobTracker
  183.      * has completely forgotten about the job.  (eg, 24 hours after the
  184.      * job completes.)
  185.      */
  186.     public NetworkedJob(JobStatus job) throws IOException {
  187.       this.status = job;
  188.       this.profile = jobSubmitClient.getJobProfile(job.getJobID());
  189.       this.statustime = System.currentTimeMillis();
  190.     }
  192.     /**
  193.      * Some methods rely on having a recent job profile object.  Refresh
  194.      * it, if necessary
  195.      */
  196.     synchronized void ensureFreshStatus() throws IOException {
  197.       if (System.currentTimeMillis() - statustime > MAX_JOBPROFILE_AGE) {
  198.         updateStatus();
  199.       }
  200.     }
  201.     
  202.     /** Some methods need to update status immediately. So, refresh
  203.      * immediately
  204.      * @throws IOException
  205.      */
  206.     synchronized void updateStatus() throws IOException {
  207.       this.status = jobSubmitClient.getJobStatus(profile.getJobID());
  208.       this.statustime = System.currentTimeMillis();
  209.     }
  211.     /**
  212.      * An identifier for the job
  213.      */
  214.     public JobID getID() {
  215.       return profile.getJobID();
  216.     }
  217.     
  218.     /** @deprecated This method is deprecated and will be removed. Applications should 
  219.      * rather use {@link #getID()}.*/
  220.     @Deprecated
  221.     public String getJobID() {
  222.       return profile.getJobID().toString();
  223.     }
  224.     
  225.     /**
  226.      * The user-specified job name
  227.      */
  228.     public String getJobName() {
  229.       return profile.getJobName();
  230.     }
  232.     /**
  233.      * The name of the job file
  234.      */
  235.     public String getJobFile() {
  236.       return profile.getJobFile();
  237.     }
  239.     /**
  240.      * A URL where the job's status can be seen
  241.      */
  242.     public String getTrackingURL() {
  243.       return profile.getURL().toString();
  244.     }
  246.     /**
  247.      * A float between 0.0 and 1.0, indicating the % of map work
  248.      * completed.
  249.      */
  250.     public float mapProgress() throws IOException {
  251.       ensureFreshStatus();
  252.       return status.mapProgress();
  253.     }
  255.     /**
  256.      * A float between 0.0 and 1.0, indicating the % of reduce work
  257.      * completed.
  258.      */
  259.     public float reduceProgress() throws IOException {
  260.       ensureFreshStatus();
  261.       return status.reduceProgress();
  262.     }
  264.     /**
  265.      * A float between 0.0 and 1.0, indicating the % of cleanup work
  266.      * completed.
  267.      */
  268.     public float cleanupProgress() throws IOException {
  269.       ensureFreshStatus();
  270.       return status.cleanupProgress();
  271.     }
  273.     /**
  274.      * A float between 0.0 and 1.0, indicating the % of setup work
  275.      * completed.
  276.      */
  277.     public float setupProgress() throws IOException {
  278.       ensureFreshStatus();
  279.       return status.setupProgress();
  280.     }
  282.     /**
  283.      * Returns immediately whether the whole job is done yet or not.
  284.      */
  285.     public synchronized boolean isComplete() throws IOException {
  286.       updateStatus();
  287.       return (status.getRunState() == JobStatus.SUCCEEDED ||
  288.               status.getRunState() == JobStatus.FAILED ||
  289.               status.getRunState() == JobStatus.KILLED);
  290.     }
  292.     /**
  293.      * True iff job completed successfully.
  294.      */
  295.     public synchronized boolean isSuccessful() throws IOException {
  296.       updateStatus();
  297.       return status.getRunState() == JobStatus.SUCCEEDED;
  298.     }
  300.     /**
  301.      * Blocks until the job is finished
  302.      */
  303.     public void waitForCompletion() throws IOException {
  304.       while (!isComplete()) {
  305.         try {
  306.           Thread.sleep(5000);
  307.         } catch (InterruptedException ie) {
  308.         }
  309.       }
  310.     }
  312.     /**
  313.      * Tells the service to get the state of the current job.
  314.      */
  315.     public synchronized int getJobState() throws IOException {
  316.       updateStatus();
  317.       return status.getRunState();
  318.     }
  319.     
  320.     /**
  321.      * Tells the service to terminate the current job.
  322.      */
  323.     public synchronized void killJob() throws IOException {
  324.       jobSubmitClient.killJob(getID());
  325.     }
  326.    
  327.     
  328.     /** Set the priority of the job.
  329.     * @param priority new priority of the job. 
  330.     */
  331.     public synchronized void setJobPriority(String priority) 
  332.                                                 throws IOException {
  333.       jobSubmitClient.setJobPriority(getID(), priority);
  334.     }
  335.     
  336.     /**
  337.      * Kill indicated task attempt.
  338.      * @param taskId the id of the task to kill.
  339.      * @param shouldFail if true the task is failed and added to failed tasks list, otherwise
  340.      * it is just killed, w/o affecting job failure status.
  341.      */
  342.     public synchronized void killTask(TaskAttemptID taskId, boolean shouldFail) throws IOException {
  343.       jobSubmitClient.killTask(taskId, shouldFail);
  344.     }
  346.     /** @deprecated Applications should rather use {@link #killTask(TaskAttemptID, boolean)}*/
  347.     @Deprecated
  348.     public synchronized void killTask(String taskId, boolean shouldFail) throws IOException {
  349.       killTask(TaskAttemptID.forName(taskId), shouldFail);
  350.     }
  351.     
  352.     /**
  353.      * Fetch task completion events from jobtracker for this job. 
  354.      */
  355.     public synchronized TaskCompletionEvent[] getTaskCompletionEvents(
  356.                                                                       int startFrom) throws IOException{
  357.       return jobSubmitClient.getTaskCompletionEvents(
  358.                                                      getID(), startFrom, 10); 
  359.     }
  361.     /**
  362.      * Dump stats to screen
  363.      */
  364.     @Override
  365.     public String toString() {
  366.       try {
  367.         updateStatus();
  368.       } catch (IOException e) {
  369.       }
  370.       return "Job: " + profile.getJobID() + "n" + 
  371.         "file: " + profile.getJobFile() + "n" + 
  372.         "tracking URL: " + profile.getURL() + "n" + 
  373.         "map() completion: " + status.mapProgress() + "n" + 
  374.         "reduce() completion: " + status.reduceProgress();
  375.     }
  376.         
  377.     /**
  378.      * Returns the counters for this job
  379.      */
  380.     public Counters getCounters() throws IOException {
  381.       return jobSubmitClient.getJobCounters(getID());
  382.     }
  383.     
  384.     @Override
  385.     public String[] getTaskDiagnostics(TaskAttemptID id) throws IOException {
  386.       return jobSubmitClient.getTaskDiagnostics(id);
  387.     }
  388.   }
  390.   private JobSubmissionProtocol jobSubmitClient;
  391.   private Path sysDir = null;
  392.   
  393.   private FileSystem fs = null;
  395.   /**
  396.    * Create a job client.
  397.    */
  398.   public JobClient() {
  399.   }
  400.     
  401.   /**
  402.    * Build a job client with the given {@link JobConf}, and connect to the 
  403.    * default {@link JobTracker}.
  404.    * 
  405.    * @param conf the job configuration.
  406.    * @throws IOException
  407.    */
  408.   public JobClient(JobConf conf) throws IOException {
  409.     setConf(conf);
  410.     init(conf);
  411.   }
  413.   /**
  414.    * Connect to the default {@link JobTracker}.
  415.    * @param conf the job configuration.
  416.    * @throws IOException
  417.    */
  418.   public void init(JobConf conf) throws IOException {
  419.     String tracker = conf.get("mapred.job.tracker", "local");
  420.     if ("local".equals(tracker)) {
  421.       this.jobSubmitClient = new LocalJobRunner(conf);
  422.     } else {
  423.       this.jobSubmitClient = createRPCProxy(JobTracker.getAddress(conf), conf);
  424.     }        
  425.   }
  427.   private JobSubmissionProtocol createRPCProxy(InetSocketAddress addr,
  428.       Configuration conf) throws IOException {
  429.     return (JobSubmissionProtocol) RPC.getProxy(JobSubmissionProtocol.class,
  430.         JobSubmissionProtocol.versionID, addr, getUGI(conf), conf,
  431.         NetUtils.getSocketFactory(conf, JobSubmissionProtocol.class));
  432.   }
  434.   /**
  435.    * Build a job client, connect to the indicated job tracker.
  436.    * 
  437.    * @param jobTrackAddr the job tracker to connect to.
  438.    * @param conf configuration.
  439.    */
  440.   public JobClient(InetSocketAddress jobTrackAddr, 
  441.                    Configuration conf) throws IOException {
  442.     jobSubmitClient = createRPCProxy(jobTrackAddr, conf);
  443.   }
  445.   /**
  446.    * Close the <code>JobClient</code>.
  447.    */
  448.   public synchronized void close() throws IOException {
  449.     if (!(jobSubmitClient instanceof LocalJobRunner)) {
  450.       RPC.stopProxy(jobSubmitClient);
  451.     }
  452.   }
  454.   /**
  455.    * Get a filesystem handle.  We need this to prepare jobs
  456.    * for submission to the MapReduce system.
  457.    * 
  458.    * @return the filesystem handle.
  459.    */
  460.   public synchronized FileSystem getFs() throws IOException {
  461.     if (this.fs == null) {
  462.       Path sysDir = getSystemDir();
  463.       this.fs = sysDir.getFileSystem(getConf());
  464.     }
  465.     return fs;
  466.   }
  467.   
  468.   /* see if two file systems are the same or not
  469.    *
  470.    */
  471.   private boolean compareFs(FileSystem srcFs, FileSystem destFs) {
  472.     URI srcUri = srcFs.getUri();
  473.     URI dstUri = destFs.getUri();
  474.     if (srcUri.getScheme() == null) {
  475.       return false;
  476.     }
  477.     if (!srcUri.getScheme().equals(dstUri.getScheme())) {
  478.       return false;
  479.     }
  480.     String srcHost = srcUri.getHost();    
  481.     String dstHost = dstUri.getHost();
  482.     if ((srcHost != null) && (dstHost != null)) {
  483.       try {
  484.         srcHost = InetAddress.getByName(srcHost).getCanonicalHostName();
  485.         dstHost = InetAddress.getByName(dstHost).getCanonicalHostName();
  486.       } catch(UnknownHostException ue) {
  487.         return false;
  488.       }
  489.       if (!srcHost.equals(dstHost)) {
  490.         return false;
  491.       }
  492.     }
  493.     else if (srcHost == null && dstHost != null) {
  494.       return false;
  495.     }
  496.     else if (srcHost != null && dstHost == null) {
  497.       return false;
  498.     }
  499.     //check for ports
  500.     if (srcUri.getPort() != dstUri.getPort()) {
  501.       return false;
  502.     }
  503.     return true;
  504.   }
  506.   // copies a file to the jobtracker filesystem and returns the path where it
  507.   // was copied to
  508.   private Path copyRemoteFiles(FileSystem jtFs, Path parentDir, Path originalPath, 
  509.                                JobConf job, short replication) throws IOException {
  510.     //check if we do not need to copy the files
  511.     // is jt using the same file system.
  512.     // just checking for uri strings... doing no dns lookups 
  513.     // to see if the filesystems are the same. This is not optimal.
  514.     // but avoids name resolution.
  515.     
  516.     FileSystem remoteFs = null;
  517.     remoteFs = originalPath.getFileSystem(job);
  518.     if (compareFs(remoteFs, jtFs)) {
  519.       return originalPath;
  520.     }
  521.     // this might have name collisions. copy will throw an exception
  522.     //parse the original path to create new path
  523.     Path newPath = new Path(parentDir, originalPath.getName());
  524.     FileUtil.copy(remoteFs, originalPath, jtFs, newPath, false, job);
  525.     jtFs.setReplication(newPath, replication);
  526.     return newPath;
  527.   }
  528.  
  529.   /**
  530.    * configure the jobconf of the user with the command line options of 
  531.    * -libjars, -files, -archives
  532.    * @param conf
  533.    * @throws IOException
  534.    */
  535.   private void configureCommandLineOptions(JobConf job, Path submitJobDir, Path submitJarFile) 
  536.     throws IOException {
  537.     
  538.     if (!(job.getBoolean("mapred.used.genericoptionsparser", false))) {
  539.       LOG.warn("Use GenericOptionsParser for parsing the arguments. " +
  540.                "Applications should implement Tool for the same.");
  541.     }
  543.     // get all the command line arguments into the 
  544.     // jobconf passed in by the user conf
  545.     String files = null;
  546.     String libjars = null;
  547.     String archives = null;
  549.     files = job.get("tmpfiles");
  550.     libjars = job.get("tmpjars");
  551.     archives = job.get("tmparchives");
  552.     /*
  553.      * set this user's id in job configuration, so later job files can be
  554.      * accessed using this user's id
  555.      */
  556.     UnixUserGroupInformation ugi = getUGI(job);
  557.       
  558.     //
  559.     // Figure out what fs the JobTracker is using.  Copy the
  560.     // job to it, under a temporary name.  This allows DFS to work,
  561.     // and under the local fs also provides UNIX-like object loading 
  562.     // semantics.  (that is, if the job file is deleted right after
  563.     // submission, we can still run the submission to completion)
  564.     //
  566.     // Create a number of filenames in the JobTracker's fs namespace
  567.     FileSystem fs = getFs();
  568.     LOG.debug("default FileSystem: " + fs.getUri());
  569.     fs.delete(submitJobDir, true);
  570.     submitJobDir = fs.makeQualified(submitJobDir);
  571.     submitJobDir = new Path(submitJobDir.toUri().getPath());
  572.     FsPermission mapredSysPerms = new FsPermission(JOB_DIR_PERMISSION);
  573.     FileSystem.mkdirs(fs, submitJobDir, mapredSysPerms);
  574.     Path filesDir = new Path(submitJobDir, "files");
  575.     Path archivesDir = new Path(submitJobDir, "archives");
  576.     Path libjarsDir = new Path(submitJobDir, "libjars");
  577.     short replication = (short)job.getInt("mapred.submit.replication", 10);
  578.     // add all the command line files/ jars and archive
  579.     // first copy them to jobtrackers filesystem 
  580.     
  581.     if (files != null) {
  582.       FileSystem.mkdirs(fs, filesDir, mapredSysPerms);
  583.       String[] fileArr = files.split(",");
  584.       for (String tmpFile: fileArr) {
  585.         Path tmp = new Path(tmpFile);
  586.         Path newPath = copyRemoteFiles(fs,filesDir, tmp, job, replication);
  587.         try {
  588.           URI pathURI = new URI(newPath.toUri().toString() + "#" + newPath.getName());
  589.           DistributedCache.addCacheFile(pathURI, job);
  590.         } catch(URISyntaxException ue) {
  591.           //should not throw a uri exception 
  592.           throw new IOException("Failed to create uri for " + tmpFile);
  593.         }
  594.         DistributedCache.createSymlink(job);
  595.       }
  596.     }
  597.     
  598.     if (libjars != null) {
  599.       FileSystem.mkdirs(fs, libjarsDir, mapredSysPerms);
  600.       String[] libjarsArr = libjars.split(",");
  601.       for (String tmpjars: libjarsArr) {
  602.         Path tmp = new Path(tmpjars);
  603.         Path newPath = copyRemoteFiles(fs, libjarsDir, tmp, job, replication);
  604.         DistributedCache.addArchiveToClassPath(newPath, job);
  605.       }
  606.     }
  607.     
  608.     
  609.     if (archives != null) {
  610.      FileSystem.mkdirs(fs, archivesDir, mapredSysPerms); 
  611.      String[] archivesArr = archives.split(",");
  612.      for (String tmpArchives: archivesArr) {
  613.        Path tmp = new Path(tmpArchives);
  614.        Path newPath = copyRemoteFiles(fs, archivesDir, tmp, job, replication);
  615.        try {
  616.          URI pathURI = new URI(newPath.toUri().toString() + "#" + newPath.getName());
  617.          DistributedCache.addCacheArchive(pathURI, job);
  618.        } catch(URISyntaxException ue) {
  619.          //should not throw an uri excpetion
  620.          throw new IOException("Failed to create uri for " + tmpArchives);
  621.        }
  622.        DistributedCache.createSymlink(job);
  623.      }
  624.     }
  625.     
  626.     //  set the timestamps of the archives and files
  627.     URI[] tarchives = DistributedCache.getCacheArchives(job);
  628.     if (tarchives != null) {
  629.       StringBuffer archiveTimestamps = 
  630.         new StringBuffer(String.valueOf(DistributedCache.getTimestamp(job, tarchives[0])));
  631.       for (int i = 1; i < tarchives.length; i++) {
  632.         archiveTimestamps.append(",");
  633.         archiveTimestamps.append(String.valueOf(DistributedCache.getTimestamp(job, tarchives[i])));
  634.       }
  635.       DistributedCache.setArchiveTimestamps(job, archiveTimestamps.toString());
  636.     }
  638.     URI[] tfiles = DistributedCache.getCacheFiles(job);
  639.     if (tfiles != null) {
  640.       StringBuffer fileTimestamps = 
  641.         new StringBuffer(String.valueOf(DistributedCache.getTimestamp(job, tfiles[0])));
  642.       for (int i = 1; i < tfiles.length; i++) {
  643.         fileTimestamps.append(",");
  644.         fileTimestamps.append(String.valueOf(DistributedCache.getTimestamp(job, tfiles[i])));
  645.       }
  646.       DistributedCache.setFileTimestamps(job, fileTimestamps.toString());
  647.     }
  648.        
  649.     String originalJarPath = job.getJar();
  651.     if (originalJarPath != null) {           // copy jar to JobTracker's fs
  652.       // use jar name if job is not named. 
  653.       if ("".equals(job.getJobName())){
  654.         job.setJobName(new Path(originalJarPath).getName());
  655.       }
  656.       job.setJar(submitJarFile.toString());
  657.       fs.copyFromLocalFile(new Path(originalJarPath), submitJarFile);
  658.       fs.setReplication(submitJarFile, replication);
  659.       fs.setPermission(submitJarFile, new FsPermission(JOB_FILE_PERMISSION));
  660.     } else {
  661.       LOG.warn("No job jar file set.  User classes may not be found. "+
  662.                "See JobConf(Class) or JobConf#setJar(String).");
  663.     }
  665.     // Set the user's name and working directory
  666.     job.setUser(ugi.getUserName());
  667.     if (ugi.getGroupNames().length > 0) {
  668.       job.set("group.name", ugi.getGroupNames()[0]);
  669.     }
  670.     if (job.getWorkingDirectory() == null) {
  671.       job.setWorkingDirectory(fs.getWorkingDirectory());          
  672.     }
  674.   }
  676.   private UnixUserGroupInformation getUGI(Configuration job) throws IOException {
  677.     UnixUserGroupInformation ugi = null;
  678.     try {
  679.       ugi = UnixUserGroupInformation.login(job, true);
  680.     } catch (LoginException e) {
  681.       throw (IOException)(new IOException(
  682.           "Failed to get the current user's information.").initCause(e));
  683.     }
  684.     return ugi;
  685.   }
  686.   
  687.   /**
  688.    * Submit a job to the MR system.
  689.    * 
  690.    * This returns a handle to the {@link RunningJob} which can be used to track
  691.    * the running-job.
  692.    * 
  693.    * @param jobFile the job configuration.
  694.    * @return a handle to the {@link RunningJob} which can be used to track the
  695.    *         running-job.
  696.    * @throws FileNotFoundException
  697.    * @throws InvalidJobConfException
  698.    * @throws IOException
  699.    */
  700.   public RunningJob submitJob(String jobFile) throws FileNotFoundException, 
  701.                                                      InvalidJobConfException, 
  702.                                                      IOException {
  703.     // Load in the submitted job details
  704.     JobConf job = new JobConf(jobFile);
  705.     return submitJob(job);
  706.   }
  707.     
  708.   // job files are world-wide readable and owner writable
  709.   final private static FsPermission JOB_FILE_PERMISSION = 
  710.     FsPermission.createImmutable((short) 0644); // rw-r--r--
  712.   // job submission directory is world readable/writable/executable
  713.   final static FsPermission JOB_DIR_PERMISSION =
  714.     FsPermission.createImmutable((short) 0777); // rwx-rwx-rwx
  715.    
  716.   /**
  717.    * Submit a job to the MR system.
  718.    * This returns a handle to the {@link RunningJob} which can be used to track
  719.    * the running-job.
  720.    * 
  721.    * @param job the job configuration.
  722.    * @return a handle to the {@link RunningJob} which can be used to track the
  723.    *         running-job.
  724.    * @throws FileNotFoundException
  725.    * @throws IOException
  726.    */
  727.   public RunningJob submitJob(JobConf job) throws FileNotFoundException,
  728.                                                   IOException {
  729.     try {
  730.       return submitJobInternal(job);
  731.     } catch (InterruptedException ie) {
  732.       throw new IOException("interrupted", ie);
  733.     } catch (ClassNotFoundException cnfe) {
  734.       throw new IOException("class not found", cnfe);
  735.     }
  736.   }
  738.   /**
  739.    * Internal method for submitting jobs to the system.
  740.    * @param job the configuration to submit
  741.    * @return a proxy object for the running job
  742.    * @throws FileNotFoundException
  743.    * @throws ClassNotFoundException
  744.    * @throws InterruptedException
  745.    * @throws IOException
  746.    */
  747.   public 
  748.   RunningJob submitJobInternal(JobConf job
  749.                                ) throws FileNotFoundException, 
  750.                                         ClassNotFoundException,
  751.                                         InterruptedException,
  752.                                         IOException {
  753.     /*
  754.      * configure the command line options correctly on the submitting dfs
  755.      */
  756.     
  757.     JobID jobId = jobSubmitClient.getNewJobId();
  758.     Path submitJobDir = new Path(getSystemDir(), jobId.toString());
  759.     Path submitJarFile = new Path(submitJobDir, "job.jar");
  760.     Path submitSplitFile = new Path(submitJobDir, "job.split");
  761.     configureCommandLineOptions(job, submitJobDir, submitJarFile);
  762.     Path submitJobFile = new Path(submitJobDir, "job.xml");
  763.     int reduces = job.getNumReduceTasks();
  764.     JobContext context = new JobContext(job, jobId);
  765.     
  766.     // Check the output specification
  767.     if (reduces == 0 ? job.getUseNewMapper() : job.getUseNewReducer()) {
  768.       org.apache.hadoop.mapreduce.OutputFormat<?,?> output =
  769.         ReflectionUtils.newInstance(context.getOutputFormatClass(), job);
  770.       output.checkOutputSpecs(context);
  771.     } else {
  772.       job.getOutputFormat().checkOutputSpecs(fs, job);
  773.     }
  775.     // Create the splits for the job
  776.     LOG.debug("Creating splits at " + fs.makeQualified(submitSplitFile));
  777.     int maps;
  778.     if (job.getUseNewMapper()) {
  779.       maps = writeNewSplits(context, submitSplitFile);
  780.     } else {
  781.       maps = writeOldSplits(job, submitSplitFile);
  782.     }
  783.     job.set("mapred.job.split.file", submitSplitFile.toString());
  784.     job.setNumMapTasks(maps);
  785.         
  786.     // Write job file to JobTracker's fs        
  787.     FSDataOutputStream out = 
  788.       FileSystem.create(fs, submitJobFile,
  789.                         new FsPermission(JOB_FILE_PERMISSION));
  791.     try {
  792.       job.writeXml(out);
  793.     } finally {
  794.       out.close();
  795.     }
  797.     //
  798.     // Now, actually submit the job (using the submit name)
  799.     //
  800.     JobStatus status = jobSubmitClient.submitJob(jobId);
  801.     if (status != null) {
  802.       return new NetworkedJob(status);
  803.     } else {
  804.       throw new IOException("Could not launch job");
  805.     }
  806.   }
  808.   private int writeOldSplits(JobConf job, 
  809.                              Path submitSplitFile) throws IOException {
  810.     InputSplit[] splits = 
  811.       job.getInputFormat().getSplits(job, job.getNumMapTasks());
  812.     // sort the splits into order based on size, so that the biggest
  813.     // go first
  814.     Arrays.sort(splits, new Comparator<InputSplit>() {
  815.       public int compare(InputSplit a, InputSplit b) {
  816.         try {
  817.           long left = a.getLength();
  818.           long right = b.getLength();
  819.           if (left == right) {
  820.             return 0;
  821.           } else if (left < right) {
  822.             return 1;
  823.           } else {
  824.             return -1;
  825.           }
  826.         } catch (IOException ie) {
  827.           throw new RuntimeException("Problem getting input split size",
  828.                                      ie);
  829.         }
  830.       }
  831.     });
  832.     DataOutputStream out = writeSplitsFileHeader(job, submitSplitFile, splits.length);
  833.     
  834.     try {
  835.       DataOutputBuffer buffer = new DataOutputBuffer();
  836.       RawSplit rawSplit = new RawSplit();
  837.       for(InputSplit split: splits) {
  838.         rawSplit.setClassName(split.getClass().getName());
  839.         buffer.reset();
  840.         split.write(buffer);
  841.         rawSplit.setDataLength(split.getLength());
  842.         rawSplit.setBytes(buffer.getData(), 0, buffer.getLength());
  843.         rawSplit.setLocations(split.getLocations());
  844.         rawSplit.write(out);
  845.       }
  846.     } finally {
  847.       out.close();
  848.     }
  849.     return splits.length;
  850.   }
  852.   private static class NewSplitComparator 
  853.     implements Comparator<org.apache.hadoop.mapreduce.InputSplit>{
  855.     @Override
  856.     public int compare(org.apache.hadoop.mapreduce.InputSplit o1,
  857.                        org.apache.hadoop.mapreduce.InputSplit o2) {
  858.       try {
  859.         long len1 = o1.getLength();
  860.         long len2 = o2.getLength();
  861.         if (len1 < len2) {
  862.           return 1;
  863.         } else if (len1 == len2) {
  864.           return 0;
  865.         } else {
  866.           return -1;
  867.         }
  868.       } catch (IOException ie) {
  869.         throw new RuntimeException("exception in compare", ie);
  870.       } catch (InterruptedException ie) {
  871.         throw new RuntimeException("exception in compare", ie);        
  872.       }
  873.     }
  874.   }
  876.   @SuppressWarnings("unchecked")
  877.   private <T extends org.apache.hadoop.mapreduce.InputSplit> 
  878.   int writeNewSplits(JobContext job, Path submitSplitFile
  879.                      ) throws IOException, InterruptedException, 
  880.                               ClassNotFoundException {
  881.     JobConf conf = job.getJobConf();
  882.     org.apache.hadoop.mapreduce.InputFormat<?,?> input =
  883.       ReflectionUtils.newInstance(job.getInputFormatClass(), job.getJobConf());
  884.     
  885.     List<org.apache.hadoop.mapreduce.InputSplit> splits = input.getSplits(job);
  886.     T[] array = (T[])
  887.       splits.toArray(new org.apache.hadoop.mapreduce.InputSplit[splits.size()]);
  889.     // sort the splits into order based on size, so that the biggest
  890.     // go first
  891.     Arrays.sort(array, new NewSplitComparator());
  892.     DataOutputStream out = writeSplitsFileHeader(conf, submitSplitFile, 
  893.                                                  array.length);
  894.     try {
  895.       if (array.length != 0) {
  896.         DataOutputBuffer buffer = new DataOutputBuffer();
  897.         RawSplit rawSplit = new RawSplit();
  898.         SerializationFactory factory = new SerializationFactory(conf);
  899.         Serializer<T> serializer = 
  900.           factory.getSerializer((Class<T>) array[0].getClass());
  901.         serializer.open(buffer);
  902.         for(T split: array) {
  903.           rawSplit.setClassName(split.getClass().getName());
  904.           buffer.reset();
  905.           serializer.serialize(split);
  906.           rawSplit.setDataLength(split.getLength());
  907.           rawSplit.setBytes(buffer.getData(), 0, buffer.getLength());
  908.           rawSplit.setLocations(split.getLocations());
  909.           rawSplit.write(out);
  910.         }
  911.         serializer.close();
  912.       }
  913.     } finally {
  914.       out.close();
  915.     }
  916.     return array.length;
  917.   }
  919.   /** 
  920.    * Checks if the job directory is clean and has all the required components 
  921.    * for (re) starting the job
  922.    */
  923.   public static boolean isJobDirValid(Path jobDirPath, FileSystem fs) 
  924.   throws IOException {
  925.     FileStatus[] contents = fs.listStatus(jobDirPath);
  926.     int matchCount = 0;
  927.     if (contents != null && contents.length >=3) {
  928.       for (FileStatus status : contents) {
  929.         if ("job.xml".equals(status.getPath().getName())) {
  930.           ++matchCount;
  931.         }
  932.         if ("job.jar".equals(status.getPath().getName())) {
  933.           ++matchCount;
  934.         }
  935.         if ("job.split".equals(status.getPath().getName())) {
  936.           ++matchCount;
  937.         }
  938.       }
  939.       if (matchCount == 3) {
  940.         return true;
  941.       }
  942.     }
  943.     return false;
  944.   }
  946.   static class RawSplit implements Writable {
  947.     private String splitClass;
  948.     private BytesWritable bytes = new BytesWritable();
  949.     private String[] locations;
  950.     long dataLength;
  952.     public void setBytes(byte[] data, int offset, int length) {
  953.       bytes.set(data, offset, length);
  954.     }
  956.     public void setClassName(String className) {
  957.       splitClass = className;
  958.     }
  959.       
  960.     public String getClassName() {
  961.       return splitClass;
  962.     }
  963.       
  964.     public BytesWritable getBytes() {
  965.       return bytes;
  966.     }
  968.     public void clearBytes() {
  969.       bytes = null;
  970.     }
  971.       
  972.     public void setLocations(String[] locations) {
  973.       this.locations = locations;
  974.     }
  975.       
  976.     public String[] getLocations() {
  977.       return locations;
  978.     }
  979.       
  980.     public void readFields(DataInput in) throws IOException {
  981.       splitClass = Text.readString(in);
  982.       dataLength = in.readLong();
  983.       bytes.readFields(in);
  984.       int len = WritableUtils.readVInt(in);
  985.       locations = new String[len];
  986.       for(int i=0; i < len; ++i) {
  987.         locations[i] = Text.readString(in);
  988.       }
  989.     }
  990.       
  991.     public void write(DataOutput out) throws IOException {
  992.       Text.writeString(out, splitClass);
  993.       out.writeLong(dataLength);
  994.       bytes.write(out);
  995.       WritableUtils.writeVInt(out, locations.length);
  996.       for(int i = 0; i < locations.length; i++) {
  997.         Text.writeString(out, locations[i]);
  998.       }        
  999.     }
  1001.     public long getDataLength() {
  1002.       return dataLength;
  1003.     }
  1004.     public void setDataLength(long l) {
  1005.       dataLength = l;
  1006.     }
  1007.     
  1008.   }
  1009.     
  1010.   private static final int CURRENT_SPLIT_FILE_VERSION = 0;
  1011.   private static final byte[] SPLIT_FILE_HEADER = "SPL".getBytes();
  1013.   private DataOutputStream writeSplitsFileHeader(Configuration conf,
  1014.                                                  Path filename,
  1015.                                                  int length
  1016.                                                  ) throws IOException {
  1017.     // write the splits to a file for the job tracker
  1018.     FileSystem fs = filename.getFileSystem(conf);
  1019.     FSDataOutputStream out = 
  1020.       FileSystem.create(fs, filename, new FsPermission(JOB_FILE_PERMISSION));
  1021.     out.write(SPLIT_FILE_HEADER);
  1022.     WritableUtils.writeVInt(out, CURRENT_SPLIT_FILE_VERSION);
  1023.     WritableUtils.writeVInt(out, length);
  1024.     return out;
  1025.   }
  1027.   /** Create the list of input splits and write them out in a file for
  1028.    *the JobTracker. The format is:
  1029.    * <format version>
  1030.    * <numSplits>
  1031.    * for each split:
  1032.    *    <RawSplit>
  1033.    * @param splits the input splits to write out
  1034.    * @param out the stream to write to
  1035.    */
  1036.   private void writeOldSplitsFile(InputSplit[] splits, 
  1037.                                   FSDataOutputStream out) throws IOException {
  1038.   }
  1040.   /**
  1041.    * Read a splits file into a list of raw splits
  1042.    * @param in the stream to read from
  1043.    * @return the complete list of splits
  1044.    * @throws IOException
  1045.    */
  1046.   static RawSplit[] readSplitFile(DataInput in) throws IOException {
  1047.     byte[] header = new byte[SPLIT_FILE_HEADER.length];
  1048.     in.readFully(header);
  1049.     if (!Arrays.equals(SPLIT_FILE_HEADER, header)) {
  1050.       throw new IOException("Invalid header on split file");
  1051.     }
  1052.     int vers = WritableUtils.readVInt(in);
  1053.     if (vers != CURRENT_SPLIT_FILE_VERSION) {
  1054.       throw new IOException("Unsupported split version " + vers);
  1055.     }
  1056.     int len = WritableUtils.readVInt(in);
  1057.     RawSplit[] result = new RawSplit[len];
  1058.     for(int i=0; i < len; ++i) {
  1059.       result[i] = new RawSplit();
  1060.       result[i].readFields(in);
  1061.     }
  1062.     return result;
  1063.   }
  1064.     
  1065.   /**
  1066.    * Get an {@link RunningJob} object to track an ongoing job.  Returns
  1067.    * null if the id does not correspond to any known job.
  1068.    * 
  1069.    * @param jobid the jobid of the job.
  1070.    * @return the {@link RunningJob} handle to track the job, null if the 
  1071.    *         <code>jobid</code> doesn't correspond to any known job.
  1072.    * @throws IOException
  1073.    */
  1074.   public RunningJob getJob(JobID jobid) throws IOException {
  1075.     JobStatus status = jobSubmitClient.getJobStatus(jobid);
  1076.     if (status != null) {
  1077.       return new NetworkedJob(status);
  1078.     } else {
  1079.       return null;
  1080.     }
  1081.   }
  1083.   /**@deprecated Applications should rather use {@link #getJob(JobID)}. 
  1084.    */
  1085.   @Deprecated
  1086.   public RunningJob getJob(String jobid) throws IOException {
  1087.     return getJob(JobID.forName(jobid));
  1088.   }
  1089.   
  1090.   /**
  1091.    * Get the information of the current state of the map tasks of a job.
  1092.    * 
  1093.    * @param jobId the job to query.
  1094.    * @return the list of all of the map tips.
  1095.    * @throws IOException
  1096.    */
  1097.   public TaskReport[] getMapTaskReports(JobID jobId) throws IOException {
  1098.     return jobSubmitClient.getMapTaskReports(jobId);
  1099.   }
  1100.   
  1101.   /**@deprecated Applications should rather use {@link #getMapTaskReports(JobID)}*/
  1102.   @Deprecated
  1103.   public TaskReport[] getMapTaskReports(String jobId) throws IOException {
  1104.     return getMapTaskReports(JobID.forName(jobId));
  1105.   }
  1106.   
  1107.   /**
  1108.    * Get the information of the current state of the reduce tasks of a job.
  1109.    * 
  1110.    * @param jobId the job to query.
  1111.    * @return the list of all of the reduce tips.
  1112.    * @throws IOException
  1113.    */    
  1114.   public TaskReport[] getReduceTaskReports(JobID jobId) throws IOException {
  1115.     return jobSubmitClient.getReduceTaskReports(jobId);
  1116.   }
  1118.   /**
  1119.    * Get the information of the current state of the cleanup tasks of a job.
  1120.    * 
  1121.    * @param jobId the job to query.
  1122.    * @return the list of all of the cleanup tips.
  1123.    * @throws IOException
  1124.    */    
  1125.   public TaskReport[] getCleanupTaskReports(JobID jobId) throws IOException {
  1126.     return jobSubmitClient.getCleanupTaskReports(jobId);
  1127.   }
  1129.   /**
  1130.    * Get the information of the current state of the setup tasks of a job.
  1131.    * 
  1132.    * @param jobId the job to query.
  1133.    * @return the list of all of the setup tips.
  1134.    * @throws IOException
  1135.    */    
  1136.   public TaskReport[] getSetupTaskReports(JobID jobId) throws IOException {
  1137.     return jobSubmitClient.getSetupTaskReports(jobId);
  1138.   }
  1140.   /**@deprecated Applications should rather use {@link #getReduceTaskReports(JobID)}*/
  1141.   @Deprecated
  1142.   public TaskReport[] getReduceTaskReports(String jobId) throws IOException {
  1143.     return getReduceTaskReports(JobID.forName(jobId));
  1144.   }
  1145.   
  1146.   /**
  1147.    * Display the information about a job's tasks, of a particular type and
  1148.    * in a particular state
  1149.    * 
  1150.    * @param jobId the ID of the job
  1151.    * @param type the type of the task (map/reduce/setup/cleanup)
  1152.    * @param state the state of the task 
  1153.    * (pending/running/completed/failed/killed)
  1154.    */
  1155.   public void displayTasks(JobID jobId, String type, String state) 
  1156.   throws IOException {
  1157.     TaskReport[] reports = new TaskReport[0];
  1158.     if (type.equals("map")) {
  1159.       reports = getMapTaskReports(jobId);
  1160.     } else if (type.equals("reduce")) {
  1161.       reports = getReduceTaskReports(jobId);
  1162.     } else if (type.equals("setup")) {
  1163.       reports = getSetupTaskReports(jobId);
  1164.     } else if (type.equals("cleanup")) {
  1165.       reports = getCleanupTaskReports(jobId);
  1166.     }
  1167.     for (TaskReport report : reports) {
  1168.       TIPStatus status = report.getCurrentStatus();
  1169.       if ((state.equals("pending") && status ==TIPStatus.PENDING) ||
  1170.           (state.equals("running") && status ==TIPStatus.RUNNING) ||
  1171.           (state.equals("completed") && status == TIPStatus.COMPLETE) ||
  1172.           (state.equals("failed") && status == TIPStatus.FAILED) ||
  1173.           (state.equals("killed") && status == TIPStatus.KILLED)) {
  1174.         printTaskAttempts(report);
  1175.       }
  1176.     }
  1177.   }
  1178.   private void printTaskAttempts(TaskReport report) {
  1179.     if (report.getCurrentStatus() == TIPStatus.COMPLETE) {
  1180.       System.out.println(report.getSuccessfulTaskAttempt());
  1181.     } else if (report.getCurrentStatus() == TIPStatus.RUNNING) {
  1182.       for (TaskAttemptID t : 
  1183.         report.getRunningTaskAttempts()) {
  1184.         System.out.println(t);
  1185.       }
  1186.     }
  1187.   }
  1188.   /**
  1189.    * Get status information about the Map-Reduce cluster.
  1190.    *  
  1191.    * @return the status information about the Map-Reduce cluster as an object
  1192.    *         of {@link ClusterStatus}.
  1193.    * @throws IOException
  1194.    */
  1195.   public ClusterStatus getClusterStatus() throws IOException {
  1196.     return getClusterStatus(false);
  1197.   }
  1199.   /**
  1200.    * Get status information about the Map-Reduce cluster.
  1201.    *  
  1202.    * @param  detailed if true then get a detailed status including the
  1203.    *         tracker names
  1204.    * @return the status information about the Map-Reduce cluster as an object
  1205.    *         of {@link ClusterStatus}.
  1206.    * @throws IOException
  1207.    */
  1208.   public ClusterStatus getClusterStatus(boolean detailed) throws IOException {
  1209.     return jobSubmitClient.getClusterStatus(detailed);
  1210.   }
  1211.     
  1213.   /** 
  1214.    * Get the jobs that are not completed and not failed.
  1215.    * 
  1216.    * @return array of {@link JobStatus} for the running/to-be-run jobs.
  1217.    * @throws IOException
  1218.    */
  1219.   public JobStatus[] jobsToComplete() throws IOException {
  1220.     return jobSubmitClient.jobsToComplete();
  1221.   }
  1223.   private static void downloadProfile(TaskCompletionEvent e
  1224.                                       ) throws IOException  {
  1225.     URLConnection connection = 
  1226.       new URL(getTaskLogURL(e.getTaskAttemptId(), e.getTaskTrackerHttp()) + 
  1227.               "&filter=profile").openConnection();
  1228.     InputStream in = connection.getInputStream();
  1229.     OutputStream out = new FileOutputStream(e.getTaskAttemptId() + ".profile");
  1230.     IOUtils.copyBytes(in, out, 64 * 1024, true);
  1231.   }
  1233.   /** 
  1234.    * Get the jobs that are submitted.
  1235.    * 
  1236.    * @return array of {@link JobStatus} for the submitted jobs.
  1237.    * @throws IOException
  1238.    */
  1239.   public JobStatus[] getAllJobs() throws IOException {
  1240.     return jobSubmitClient.getAllJobs();
  1241.   }
  1242.   
  1243.   /** 
  1244.    * Utility that submits a job, then polls for progress until the job is
  1245.    * complete.
  1246.    * 
  1247.    * @param job the job configuration.
  1248.    * @throws IOException if the job fails
  1249.    */
  1250.   public static RunningJob runJob(JobConf job) throws IOException {
  1251.     JobClient jc = new JobClient(job);
  1252.     RunningJob rj = jc.submitJob(job);
  1253.     try {
  1254.       if (!jc.monitorAndPrintJob(job, rj)) {
  1255.         throw new IOException("Job failed!");
  1256.       }
  1257.     } catch (InterruptedException ie) {
  1258.       Thread.currentThread().interrupt();
  1259.     }
  1260.     return rj;
  1261.   }
  1262.   
  1263.   /**
  1264.    * Monitor a job and print status in real-time as progress is made and tasks 
  1265.    * fail.
  1266.    * @param conf the job's configuration
  1267.    * @param job the job to track
  1268.    * @return true if the job succeeded
  1269.    * @throws IOException if communication to the JobTracker fails
  1270.    */
  1271.   public boolean monitorAndPrintJob(JobConf conf, 
  1272.                                     RunningJob job
  1273.   ) throws IOException, InterruptedException {
  1274.     String lastReport = null;
  1275.     TaskStatusFilter filter;
  1276.     filter = getTaskOutputFilter(conf);
  1277.     JobID jobId = job.getID();
  1278.     LOG.info("Running job: " + jobId);
  1279.     int eventCounter = 0;
  1280.     boolean profiling = conf.getProfileEnabled();
  1281.     Configuration.IntegerRanges mapRanges = conf.getProfileTaskRange(true);
  1282.     Configuration.IntegerRanges reduceRanges = conf.getProfileTaskRange(false);
  1284.     while (!job.isComplete()) {
  1285.       Thread.sleep(1000);
  1286.       String report = 
  1287.         (" map " + StringUtils.formatPercent(job.mapProgress(), 0)+
  1288.             " reduce " + 
  1289.             StringUtils.formatPercent(job.reduceProgress(), 0));
  1290.       if (!report.equals(lastReport)) {
  1291.         LOG.info(report);
  1292.         lastReport = report;
  1293.       }
  1295.       TaskCompletionEvent[] events = 
  1296.         job.getTaskCompletionEvents(eventCounter); 
  1297.       eventCounter += events.length;
  1298.       for(TaskCompletionEvent event : events){
  1299.         TaskCompletionEvent.Status status = event.getTaskStatus();
  1300.         if (profiling && 
  1301.             (status == TaskCompletionEvent.Status.SUCCEEDED ||
  1302.                 status == TaskCompletionEvent.Status.FAILED) &&
  1303.                 (event.isMap ? mapRanges : reduceRanges).
  1304.                 isIncluded(event.idWithinJob())) {
  1305.           downloadProfile(event);
  1306.         }
  1307.         switch(filter){
  1308.         case NONE:
  1309.           break;
  1310.         case SUCCEEDED:
  1311.           if (event.getTaskStatus() == 
  1312.             TaskCompletionEvent.Status.SUCCEEDED){
  1313.             LOG.info(event.toString());
  1314.             displayTaskLogs(event.getTaskAttemptId(), event.getTaskTrackerHttp());
  1315.           }
  1316.           break; 
  1317.         case FAILED:
  1318.           if (event.getTaskStatus() == 
  1319.             TaskCompletionEvent.Status.FAILED){
  1320.             LOG.info(event.toString());
  1321.             // Displaying the task diagnostic information
  1322.             TaskAttemptID taskId = event.getTaskAttemptId();
  1323.             String[] taskDiagnostics = 
  1324.               jobSubmitClient.getTaskDiagnostics(taskId); 
  1325.             if (taskDiagnostics != null) {
  1326.               for(String diagnostics : taskDiagnostics){
  1327.                 System.err.println(diagnostics);
  1328.               }
  1329.             }
  1330.             // Displaying the task logs
  1331.             displayTaskLogs(event.getTaskAttemptId(), event.getTaskTrackerHttp());
  1332.           }
  1333.           break; 
  1334.         case KILLED:
  1335.           if (event.getTaskStatus() == TaskCompletionEvent.Status.KILLED){
  1336.             LOG.info(event.toString());
  1337.           }
  1338.           break; 
  1339.         case ALL:
  1340.           LOG.info(event.toString());
  1341.           displayTaskLogs(event.getTaskAttemptId(), event.getTaskTrackerHttp());
  1342.           break;
  1343.         }
  1344.       }
  1345.     }
  1346.     LOG.info("Job complete: " + jobId);
  1347.     job.getCounters().log(LOG);
  1348.     return job.isSuccessful();
  1349.   }
  1351.   static String getTaskLogURL(TaskAttemptID taskId, String baseUrl) {
  1352.     return (baseUrl + "/tasklog?plaintext=true&taskid=" + taskId); 
  1353.   }
  1354.   
  1355.   private static void displayTaskLogs(TaskAttemptID taskId, String baseUrl)
  1356.     throws IOException {
  1357.     // The tasktracker for a 'failed/killed' job might not be around...
  1358.     if (baseUrl != null) {
  1359.       // Construct the url for the tasklogs
  1360.       String taskLogUrl = getTaskLogURL(taskId, baseUrl);
  1361.       
  1362.       // Copy tasks's stdout of the JobClient
  1363.       getTaskLogs(taskId, new URL(taskLogUrl+"&filter=stdout"), System.out);
  1364.         
  1365.       // Copy task's stderr to stderr of the JobClient 
  1366.       getTaskLogs(taskId, new URL(taskLogUrl+"&filter=stderr"), System.err);
  1367.     }
  1368.   }
  1369.     
  1370.   private static void getTaskLogs(TaskAttemptID taskId, URL taskLogUrl, 
  1371.                                   OutputStream out) {
  1372.     try {
  1373.       URLConnection connection = taskLogUrl.openConnection();
  1374.       BufferedReader input = 
  1375.         new BufferedReader(new InputStreamReader(connection.getInputStream()));
  1376.       BufferedWriter output = 
  1377.         new BufferedWriter(new OutputStreamWriter(out));
  1378.       try {
  1379.         String logData = null;
  1380.         while ((logData = input.readLine()) != null) {
  1381.           if (logData.length() > 0) {
  1382.             output.write(taskId + ": " + logData + "n");
  1383.             output.flush();
  1384.           }
  1385.         }
  1386.       } finally {
  1387.         input.close();
  1388.       }
  1389.     }catch(IOException ioe){
  1390.       LOG.warn("Error reading task output" + ioe.getMessage()); 
  1391.     }
  1392.   }    
  1394.   static Configuration getConfiguration(String jobTrackerSpec)
  1395.   {
  1396.     Configuration conf = new Configuration();
  1397.     if (jobTrackerSpec != null) {        
  1398.       if (jobTrackerSpec.indexOf(":") >= 0) {
  1399.         conf.set("mapred.job.tracker", jobTrackerSpec);
  1400.       } else {
  1401.         String classpathFile = "hadoop-" + jobTrackerSpec + ".xml";
  1402.         URL validate = conf.getResource(classpathFile);
  1403.         if (validate == null) {
  1404.           throw new RuntimeException(classpathFile + " not found on CLASSPATH");
  1405.         }
  1406.         conf.addResource(classpathFile);
  1407.       }
  1408.     }
  1409.     return conf;
  1410.   }
  1412.   /**
  1413.    * Sets the output filter for tasks. only those tasks are printed whose
  1414.    * output matches the filter. 
  1415.    * @param newValue task filter.
  1416.    */
  1417.   @Deprecated
  1418.   public void setTaskOutputFilter(TaskStatusFilter newValue){
  1419.     this.taskOutputFilter = newValue;
  1420.   }
  1421.     
  1422.   /**
  1423.    * Get the task output filter out of the JobConf.
  1424.    * 
  1425.    * @param job the JobConf to examine.
  1426.    * @return the filter level.
  1427.    */
  1428.   public static TaskStatusFilter getTaskOutputFilter(JobConf job) {
  1429.     return TaskStatusFilter.valueOf(job.get("jobclient.output.filter", 
  1430.                                             "FAILED"));
  1431.   }
  1432.     
  1433.   /**
  1434.    * Modify the JobConf to set the task output filter.
  1435.    * 
  1436.    * @param job the JobConf to modify.
  1437.    * @param newValue the value to set.
  1438.    */
  1439.   public static void setTaskOutputFilter(JobConf job, 
  1440.                                          TaskStatusFilter newValue) {
  1441.     job.set("jobclient.output.filter", newValue.toString());
  1442.   }
  1443.     
  1444.   /**
  1445.    * Returns task output filter.
  1446.    * @return task filter. 
  1447.    */
  1448.   @Deprecated
  1449.   public TaskStatusFilter getTaskOutputFilter(){
  1450.     return this.taskOutputFilter; 
  1451.   }
  1453.   private String getJobPriorityNames() {
  1454.     StringBuffer sb = new StringBuffer();
  1455.     for (JobPriority p : JobPriority.values()) {
  1456.       sb.append(p.name()).append(" ");
  1457.     }
  1458.     return sb.substring(0, sb.length()-1);
  1459.   }
  1460.   
  1461.   /**
  1462.    * Display usage of the command-line tool and terminate execution
  1463.    */
  1464.   private void displayUsage(String cmd) {
  1465.     String prefix = "Usage: JobClient ";
  1466.     String jobPriorityValues = getJobPriorityNames();
  1467.     String taskTypes = "map, reduce, setup, cleanup";
  1468.     String taskStates = "running, completed";
  1469.     if("-submit".equals(cmd)) {
  1470.       System.err.println(prefix + "[" + cmd + " <job-file>]");
  1471.     } else if ("-status".equals(cmd) || "-kill".equals(cmd)) {
  1472.       System.err.println(prefix + "[" + cmd + " <job-id>]");
  1473.     } else if ("-counter".equals(cmd)) {
  1474.       System.err.println(prefix + "[" + cmd + " <job-id> <group-name> <counter-name>]");
  1475.     } else if ("-events".equals(cmd)) {
  1476.       System.err.println(prefix + "[" + cmd + " <job-id> <from-event-#> <#-of-events>]");
  1477.     } else if ("-history".equals(cmd)) {
  1478.       System.err.println(prefix + "[" + cmd + " <jobOutputDir>]");
  1479.     } else if ("-list".equals(cmd)) {
  1480.       System.err.println(prefix + "[" + cmd + " [all]]");
  1481.     } else if ("-kill-task".equals(cmd) || "-fail-task".equals(cmd)) {
  1482.       System.err.println(prefix + "[" + cmd + " <task-id>]");
  1483.     } else if ("-set-priority".equals(cmd)) {
  1484.       System.err.println(prefix + "[" + cmd + " <job-id> <priority>]. " +
  1485.           "Valid values for priorities are: " 
  1486.           + jobPriorityValues); 
  1487.     } else if ("-list-active-trackers".equals(cmd)) {
  1488.       System.err.println(prefix + "[" + cmd + "]");
  1489.     } else if ("-list-blacklisted-trackers".equals(cmd)) {
  1490.       System.err.println(prefix + "[" + cmd + "]");
  1491.     } else if ("-list-attempt-ids".equals(cmd)) {
  1492.       System.err.println(prefix + "[" + cmd + 
  1493.           " <job-id> <task-type> <task-state>]. " +
  1494.           "Valid values for <task-type> are " + taskTypes + ". " +
  1495.           "Valid values for <task-state> are " + taskStates);
  1496.     } else {
  1497.       System.err.printf(prefix + "<command> <args>n");
  1498.       System.err.printf("t[-submit <job-file>]n");
  1499.       System.err.printf("t[-status <job-id>]n");
  1500.       System.err.printf("t[-counter <job-id> <group-name> <counter-name>]n");
  1501.       System.err.printf("t[-kill <job-id>]n");
  1502.       System.err.printf("t[-set-priority <job-id> <priority>]. " +
  1503.                                       "Valid values for priorities are: " +
  1504.                                       jobPriorityValues + "n");
  1505.       System.err.printf("t[-events <job-id> <from-event-#> <#-of-events>]n");
  1506.       System.err.printf("t[-history <jobOutputDir>]n");
  1507.       System.err.printf("t[-list [all]]n");
  1508.       System.err.printf("t[-list-active-trackers]n");
  1509.       System.err.printf("t[-list-blacklisted-trackers]n");
  1510.       System.err.println("t[-list-attempt-ids <job-id> <task-type> " +
  1511.        "<task-state>]n");
  1512.       System.err.printf("t[-kill-task <task-id>]n");
  1513.       System.err.printf("t[-fail-task <task-id>]nn");
  1514.       ToolRunner.printGenericCommandUsage(System.out);
  1515.     }
  1516.   }
  1517.     
  1518.   public int run(String[] argv) throws Exception {
  1519.     int exitCode = -1;
  1520.     if (argv.length < 1) {
  1521.       displayUsage("");
  1522.       return exitCode;
  1523.     }    
  1524.     // process arguments
  1525.     String cmd = argv[0];
  1526.     String submitJobFile = null;
  1527.     String jobid = null;
  1528.     String taskid = null;
  1529.     String outputDir = null;
  1530.     String counterGroupName = null;
  1531.     String counterName = null;
  1532.     String newPriority = null;
  1533.     String taskType = null;
  1534.     String taskState = null;
  1535.     int fromEvent = 0;
  1536.     int nEvents = 0;
  1537.     boolean getStatus = false;
  1538.     boolean getCounter = false;
  1539.     boolean killJob = false;
  1540.     boolean listEvents = false;
  1541.     boolean viewHistory = false;
  1542.     boolean viewAllHistory = false;
  1543.     boolean listJobs = false;
  1544.     boolean listAllJobs = false;
  1545.     boolean listActiveTrackers = false;
  1546.     boolean listBlacklistedTrackers = false;
  1547.     boolean displayTasks = false;
  1548.     boolean killTask = false;
  1549.     boolean failTask = false;
  1550.     boolean setJobPriority = false;
  1552.     if ("-submit".equals(cmd)) {
  1553.       if (argv.length != 2) {
  1554.         displayUsage(cmd);
  1555.         return exitCode;
  1556.       }
  1557.       submitJobFile = argv[1];
  1558.     } else if ("-status".equals(cmd)) {
  1559.       if (argv.length != 2) {
  1560.         displayUsage(cmd);
  1561.         return exitCode;
  1562.       }
  1563.       jobid = argv[1];
  1564.       getStatus = true;
  1565.     } else if("-counter".equals(cmd)) {
  1566.       if (argv.length != 4) {
  1567.         displayUsage(cmd);
  1568.         return exitCode;
  1569.       }
  1570.       getCounter = true;
  1571.       jobid = argv[1];
  1572.       counterGroupName = argv[2];
  1573.       counterName = argv[3];
  1574.     } else if ("-kill".equals(cmd)) {
  1575.       if (argv.length != 2) {
  1576.         displayUsage(cmd);
  1577.         return exitCode;
  1578.       }
  1579.       jobid = argv[1];
  1580.       killJob = true;
  1581.     } else if ("-set-priority".equals(cmd)) {
  1582.       if (argv.length != 3) {
  1583.         displayUsage(cmd);
  1584.         return exitCode;
  1585.       }
  1586.       jobid = argv[1];
  1587.       newPriority = argv[2];
  1588.       try {
  1589.         JobPriority jp = JobPriority.valueOf(newPriority); 
  1590.       } catch (IllegalArgumentException iae) {
  1591.         displayUsage(cmd);
  1592.         return exitCode;
  1593.       }
  1594.       setJobPriority = true; 
  1595.     } else if ("-events".equals(cmd)) {
  1596.       if (argv.length != 4) {
  1597.         displayUsage(cmd);
  1598.         return exitCode;
  1599.       }
  1600.       jobid = argv[1];
  1601.       fromEvent = Integer.parseInt(argv[2]);
  1602.       nEvents = Integer.parseInt(argv[3]);
  1603.       listEvents = true;
  1604.     } else if ("-history".equals(cmd)) {
  1605.       if (argv.length != 2 && !(argv.length == 3 && "all".equals(argv[1]))) {
  1606.          displayUsage(cmd);
  1607.          return exitCode;
  1608.       }
  1609.       viewHistory = true;
  1610.       if (argv.length == 3 && "all".equals(argv[1])) {
  1611.          viewAllHistory = true;
  1612.          outputDir = argv[2];
  1613.       } else {
  1614.          outputDir = argv[1];
  1615.       }
  1616.     } else if ("-list".equals(cmd)) {
  1617.       if (argv.length != 1 && !(argv.length == 2 && "all".equals(argv[1]))) {
  1618.         displayUsage(cmd);
  1619.         return exitCode;
  1620.       }
  1621.       if (argv.length == 2 && "all".equals(argv[1])) {
  1622.         listAllJobs = true;
  1623.       } else {
  1624.         listJobs = true;
  1625.       }
  1626.     } else if("-kill-task".equals(cmd)) {
  1627.       if(argv.length != 2) {
  1628.         displayUsage(cmd);
  1629.         return exitCode;
  1630.       }
  1631.       killTask = true;
  1632.       taskid = argv[1];
  1633.     } else if("-fail-task".equals(cmd)) {
  1634.       if(argv.length != 2) {
  1635.         displayUsage(cmd);
  1636.         return exitCode;
  1637.       }
  1638.       failTask = true;
  1639.       taskid = argv[1];
  1640.     } else if ("-list-active-trackers".equals(cmd)) {
  1641.       if (argv.length != 1) {
  1642.         displayUsage(cmd);
  1643.         return exitCode;
  1644.       }
  1645.       listActiveTrackers = true;
  1646.     } else if ("-list-blacklisted-trackers".equals(cmd)) {
  1647.       if (argv.length != 1) {
  1648.         displayUsage(cmd);
  1649.         return exitCode;
  1650.       }
  1651.       listBlacklistedTrackers = true;
  1652.     } else if ("-list-attempt-ids".equals(cmd)) {
  1653.       if (argv.length != 4) {
  1654.         displayUsage(cmd);
  1655.         return exitCode;
  1656.       }
  1657.       jobid = argv[1];
  1658.       taskType = argv[2];
  1659.       taskState = argv[3];
  1660.       displayTasks = true;
  1661.     } else {
  1662.       displayUsage(cmd);
  1663.       return exitCode;
  1664.     }
  1666.     // initialize JobClient
  1667.     JobConf conf = null;
  1668.     if (submitJobFile != null) {
  1669.       conf = new JobConf(submitJobFile);
  1670.     } else {
  1671.       conf = new JobConf(getConf());
  1672.     }
  1673.     init(conf);
  1674.         
  1675.     // Submit the request
  1676.     try {
  1677.       if (submitJobFile != null) {
  1678.         RunningJob job = submitJob(conf);
  1679.         System.out.println("Created job " + job.getID());
  1680.         exitCode = 0;
  1681.       } else if (getStatus) {
  1682.         RunningJob job = getJob(JobID.forName(jobid));
  1683.         if (job == null) {
  1684.           System.out.println("Could not find job " + jobid);
  1685.         } else {
  1686.           System.out.println();
  1687.           System.out.println(job);
  1688.           System.out.println(job.getCounters());
  1689.           exitCode = 0;
  1690.         }
  1691.       } else if (getCounter) {
  1692.         RunningJob job = getJob(JobID.forName(jobid));
  1693.         if (job == null) {
  1694.           System.out.println("Could not find job " + jobid);
  1695.         } else {
  1696.           Counters counters = job.getCounters();
  1697.           Group group = counters.getGroup(counterGroupName);
  1698.           Counter counter = group.getCounterForName(counterName);
  1699.           System.out.println(counter.getCounter());
  1700.           exitCode = 0;
  1701.         }
  1702.       } else if (killJob) {
  1703.         RunningJob job = getJob(JobID.forName(jobid));
  1704.         if (job == null) {
  1705.           System.out.println("Could not find job " + jobid);
  1706.         } else {
  1707.           job.killJob();
  1708.           System.out.println("Killed job " + jobid);
  1709.           exitCode = 0;
  1710.         }
  1711.       } else if (setJobPriority) {
  1712.         RunningJob job = getJob(JobID.forName(jobid));
  1713.         if (job == null) {
  1714.           System.out.println("Could not find job " + jobid);
  1715.         } else {
  1716.           job.setJobPriority(newPriority);
  1717.           System.out.println("Changed job priority.");
  1718.           exitCode = 0;
  1719.         } 
  1720.       } else if (viewHistory) {
  1721.         viewHistory(outputDir, viewAllHistory);
  1722.         exitCode = 0;
  1723.       } else if (listEvents) {
  1724.         listEvents(JobID.forName(jobid), fromEvent, nEvents);
  1725.         exitCode = 0;
  1726.       } else if (listJobs) {
  1727.         listJobs();
  1728.         exitCode = 0;
  1729.       } else if (listAllJobs) {
  1730.         listAllJobs();
  1731.         exitCode = 0;
  1732.       } else if (listActiveTrackers) {
  1733.         listActiveTrackers();
  1734.         exitCode = 0;
  1735.       } else if (listBlacklistedTrackers) {
  1736.         listBlacklistedTrackers();
  1737.         exitCode = 0;
  1738.       } else if (displayTasks) {
  1739.         displayTasks(JobID.forName(jobid), taskType, taskState);
  1740.       } else if(killTask) {
  1741.         if(jobSubmitClient.killTask(TaskAttemptID.forName(taskid), false)) {
  1742.           System.out.println("Killed task " + taskid);
  1743.           exitCode = 0;
  1744.         } else {
  1745.           System.out.println("Could not kill task " + taskid);
  1746.           exitCode = -1;
  1747.         }
  1748.       } else if(failTask) {
  1749.         if(jobSubmitClient.killTask(TaskAttemptID.forName(taskid), true)) {
  1750.           System.out.println("Killed task " + taskid + " by failing it");
  1751.           exitCode = 0;
  1752.         } else {
  1753.           System.out.println("Could not fail task " + taskid);
  1754.           exitCode = -1;
  1755.         }
  1756.       }
  1757.     } finally {
  1758.       close();
  1759.     }
  1760.     return exitCode;
  1761.   }
  1763.   private void viewHistory(String outputDir, boolean all) 
  1764.     throws IOException {
  1765.     HistoryViewer historyViewer = new HistoryViewer(outputDir,
  1766.                                         getConf(), all);
  1767.     historyViewer.print();
  1768.   }
  1769.   
  1770.   /**
  1771.    * List the events for the given job
  1772.    * @param jobId the job id for the job's events to list
  1773.    * @throws IOException
  1774.    */
  1775.   private void listEvents(JobID jobId, int fromEventId, int numEvents)
  1776.     throws IOException {
  1777.     TaskCompletionEvent[] events = 
  1778.       jobSubmitClient.getTaskCompletionEvents(jobId, fromEventId, numEvents);
  1779.     System.out.println("Task completion events for " + jobId);
  1780.     System.out.println("Number of events (from " + fromEventId + 
  1781.                        ") are: " + events.length);
  1782.     for(TaskCompletionEvent event: events) {
  1783.       System.out.println(event.getTaskStatus() + " " + event.getTaskAttemptId() + " " + 
  1784.                          getTaskLogURL(event.getTaskAttemptId(), 
  1785.                                        event.getTaskTrackerHttp()));
  1786.     }
  1787.   }
  1789.   /**
  1790.    * Dump a list of currently running jobs
  1791.    * @throws IOException
  1792.    */
  1793.   private void listJobs() throws IOException {
  1794.     JobStatus[] jobs = jobsToComplete();
  1795.     if (jobs == null)
  1796.       jobs = new JobStatus[0];
  1798.     System.out.printf("%d jobs currently runningn", jobs.length);
  1799.     displayJobList(jobs);
  1800.   }
  1801.     
  1802.   /**
  1803.    * Dump a list of all jobs submitted.
  1804.    * @throws IOException
  1805.    */
  1806.   private void listAllJobs() throws IOException {
  1807.     JobStatus[] jobs = getAllJobs();
  1808.     if (jobs == null)
  1809.       jobs = new JobStatus[0];
  1810.     System.out.printf("%d jobs submittedn", jobs.length);
  1811.     System.out.printf("States are:ntRunning : 1tSucceded : 2" +
  1812.     "tFailed : 3tPrep : 4n");
  1813.     displayJobList(jobs);
  1814.   }
  1815.   
  1816.   /**
  1817.    * Display the list of active trackers
  1818.    */
  1819.   private void listActiveTrackers() throws IOException {
  1820.     ClusterStatus c = jobSubmitClient.getClusterStatus(true);
  1821.     Collection<String> trackers = c.getActiveTrackerNames();
  1822.     for (String trackerName : trackers) {
  1823.       System.out.println(trackerName);
  1824.     }
  1825.   }
  1827.   /**
  1828.    * Display the list of blacklisted trackers
  1829.    */
  1830.   private void listBlacklistedTrackers() throws IOException {
  1831.     ClusterStatus c = jobSubmitClient.getClusterStatus(true);
  1832.     Collection<String> trackers = c.getBlacklistedTrackerNames();
  1833.     for (String trackerName : trackers) {
  1834.       System.out.println(trackerName);
  1835.     }
  1836.   }
  1838.   void displayJobList(JobStatus[] jobs) {
  1839.     System.out.printf("JobIdtStatetStartTimetUserNametPrioritytSchedulingInfon");
  1840.     for (JobStatus job : jobs) {
  1841.       System.out.printf("%st%dt%dt%st%st%sn", job.getJobID(), job.getRunState(),
  1842.           job.getStartTime(), job.getUsername(), 
  1843.           job.getJobPriority().name(), job.getSchedulingInfo());
  1844.     }
  1845.   }
  1847.   /**
  1848.    * Get status information about the max available Maps in the cluster.
  1849.    *  
  1850.    * @return the max available Maps in the cluster
  1851.    * @throws IOException
  1852.    */
  1853.   public int getDefaultMaps() throws IOException {
  1854.     return getClusterStatus().getMaxMapTasks();
  1855.   }
  1857.   /**
  1858.    * Get status information about the max available Reduces in the cluster.
  1859.    *  
  1860.    * @return the max available Reduces in the cluster
  1861.    * @throws IOException
  1862.    */
  1863.   public int getDefaultReduces() throws IOException {
  1864.     return getClusterStatus().getMaxReduceTasks();
  1865.   }
  1867.   /**
  1868.    * Grab the jobtracker system directory path where job-specific files are to be placed.
  1869.    * 
  1870.    * @return the system directory where job-specific files are to be placed.
  1871.    */
  1872.   public Path getSystemDir() {
  1873.     if (sysDir == null) {
  1874.       sysDir = new Path(jobSubmitClient.getSystemDir());
  1875.     }
  1876.     return sysDir;
  1877.   }
  1878.   
  1879.   
  1880.   /**
  1881.    * Return an array of queue information objects about all the Job Queues
  1882.    * configured.
  1883.    * 
  1884.    * @return Array of JobQueueInfo objects
  1885.    * @throws IOException
  1886.    */
  1887.   public JobQueueInfo[] getQueues() throws IOException {
  1888.     return jobSubmitClient.getQueues();
  1889.   }
  1890.   
  1891.   /**
  1892.    * Gets all the jobs which were added to particular Job Queue
  1893.    * 
  1894.    * @param queueName name of the Job Queue
  1895.    * @return Array of jobs present in the job queue
  1896.    * @throws IOException
  1897.    */
  1898.   
  1899.   public JobStatus[] getJobsFromQueue(String queueName) throws IOException {
  1900.     return jobSubmitClient.getJobsFromQueue(queueName);
  1901.   }
  1902.   
  1903.   /**
  1904.    * Gets the queue information associated to a particular Job Queue
  1905.    * 
  1906.    * @param queueName name of the job queue.
  1907.    * @return Queue information associated to particular queue.
  1908.    * @throws IOException
  1909.    */
  1910.   public JobQueueInfo getQueueInfo(String queueName) throws IOException {
  1911.     return jobSubmitClient.getQueueInfo(queueName);
  1912.   }
  1913.   
  1915.   /**
  1916.    */
  1917.   public static void main(String argv[]) throws Exception {
  1918.     int res = ToolRunner.run(new JobClient(), argv);
  1919.     System.exit(res);
  1920.   }
  1921. }