View Javadoc

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  
19  package org.apache.hadoop.hbase.mapreduce;
20  
21  import java.io.IOException;
22  import java.lang.reflect.Constructor;
23  import java.lang.reflect.InvocationTargetException;
24  import java.lang.reflect.Method;
25  
26  import org.apache.commons.logging.Log;
27  import org.apache.commons.logging.LogFactory;
28  import org.apache.hadoop.hbase.classification.InterfaceAudience;
29  import org.apache.hadoop.hbase.classification.InterfaceStability;
30  import org.apache.hadoop.fs.Path;
31  import org.apache.hadoop.conf.Configuration;
32  import org.apache.hadoop.mapred.JobClient;
33  import org.apache.hadoop.mapred.JobConf;
34  import org.apache.hadoop.mapreduce.JobSubmissionFiles;
35  
36  /**
37   * Utility methods to interact with a job.
38   */
39  @InterfaceAudience.Private
40  @InterfaceStability.Evolving
41  public abstract class JobUtil {
42    private static final Log LOG = LogFactory.getLog(JobUtil.class);
43  
44    protected JobUtil() {
45      super();
46    }
47  
48    /**
49     * Initializes the staging directory and returns the path.
50     * <p>
51     * CLOUDERA-SPECIFIC-NOTE:
52     * MR1 and MR2 are incompatible regarding getStagingDir() API.
53     * <p>
54     * The following code is to handle both MR1 and MR2 at
55     * compile/run time. This is done using reflection to figure out the right API.
56     *
57     * @param conf system configuration
58     * @return staging directory path
59     * @throws IOException
60     * @throws InterruptedException
61     */
62    public static Path getStagingDir(Configuration conf)
63        throws IOException, InterruptedException {
64      Path stagingDirPath = null;
65      // The API to get staging directory is different in MR1 and MR2/YARN. We first try MR1, and if
66      // it is not present, fall back to MR2.
67      try {
68        stagingDirPath = getStagingDirFromMR1(conf);
69      } catch (NoSuchMethodException e) {
70        stagingDirPath = getStagingDirFromMR2(conf);
71      }
72      if (stagingDirPath != null) LOG.debug("Staging dir of the job is: " + stagingDirPath);
73      return stagingDirPath;
74    }
75  
76    /**
77     * Invokes {@link JobSubmissionFiles#getStagingDir(org.apache.hadoop.mapred.JobClient,
78     * Configuration)} API, if present, to get staging dir path.
79     * @param conf
80     * @return stagingDir path for the job
81     * @throws IOException
82     * @throws NoSuchMethodException
83     */
84    private static Path getStagingDirFromMR1(Configuration conf) throws IOException,
85        NoSuchMethodException, InterruptedException {
86      Path stagingDirPath;
87      JobClient jobClient = new JobClient(new JobConf(conf));
88      Method getStagingDirMethod = JobSubmissionFiles.class.getMethod("getStagingDir",
89        jobClient.getClass(), conf.getClass());
90      try {
91        // call this mr1 specific call:
92        // JobSubmissionFiles.getStagingDir(jobClient, conf);
93        stagingDirPath = (Path) getStagingDirMethod.invoke(null, jobClient, conf);
94      } catch (IllegalArgumentException iae) {
95        throw new IllegalStateException(iae);
96      } catch (IllegalAccessException e) {
97        throw new IllegalStateException(e);
98      } catch (InvocationTargetException ite) {
99        throw new IllegalStateException(ite);
100     }
101     return stagingDirPath;
102   }
103 
104   /**
105    * Invokes {@link JobSubmissionFiles#getStagingDir(org.apache.hadoop.mapreduce.Cluster,
106    *  Configuration)} API, if present, to get the staging dir path.
107    * @param conf
108    * @return stagingDir path for the job
109    */
110   private static Path getStagingDirFromMR2(Configuration conf) {
111     Path stagingDirPath = null;
112     try {
113       Class<?> clusterClass = Class.forName("org.apache.hadoop.mapreduce.Cluster");
114       Method getStagingDirMethod = JobSubmissionFiles.class.getMethod("getStagingDir",
115         clusterClass, conf.getClass());
116       Constructor<?> ctr = clusterClass.getConstructor(conf.getClass());
117       Object clusterInstance = ctr.newInstance(conf);
118       // call this mr2 specific call:
119       // JobSubmissionFiles.getStagingDir(cluster, conf);
120       stagingDirPath = (Path) getStagingDirMethod.invoke(null, clusterInstance, conf);
121     } catch (ClassNotFoundException cnfe) {
122       throw new IllegalStateException(cnfe);
123     } catch (SecurityException se) {
124       throw new IllegalStateException(se);
125     } catch (NoSuchMethodException nsme) {
126       throw new IllegalStateException(nsme);
127     } catch (IllegalArgumentException iae) {
128       throw new IllegalStateException(iae);
129     } catch (InstantiationException ie) {
130       throw new IllegalStateException(ie);
131     } catch (IllegalAccessException e) {
132       throw new IllegalStateException(e);
133     } catch (InvocationTargetException ite) {
134       throw new IllegalStateException(ite);
135     }
136     return stagingDirPath;
137   }
138 }