Source code

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 */
018
019package org.apache.oozie.client;
020
021import java.io.BufferedReader;
022import java.io.File;
023import java.io.FileReader;
024import java.io.IOException;
025import java.io.InputStreamReader;
026import java.net.HttpURLConnection;
027import java.util.Properties;
028
029import org.apache.oozie.cli.OozieCLI;
030import org.apache.oozie.client.rest.JsonTags;
031import org.apache.oozie.client.rest.RestConstants;
032import org.json.simple.JSONObject;
033import org.json.simple.JSONValue;
034
035public class XOozieClient extends OozieClient {
036
037    public static final String JT = "mapred.job.tracker";
038    public static final String JT_2 = "mapreduce.jobtracker.address";
039
040    public static final String NN = "fs.default.name";
041    public static final String NN_2 = "fs.defaultFS";
042
043    @Deprecated
044    public static final String JT_PRINCIPAL = "mapreduce.jobtracker.kerberos.principal";
045
046    @Deprecated
047    public static final String NN_PRINCIPAL = "dfs.namenode.kerberos.principal";
048
049    public static final String PIG_SCRIPT = "oozie.pig.script";
050
051    public static final String PIG_OPTIONS = "oozie.pig.options";
052
053    public static final String PIG_SCRIPT_PARAMS = "oozie.pig.script.params";
054
055    public static final String HIVE_SCRIPT = "oozie.hive.script";
056
057    public static final String HIVE_OPTIONS = "oozie.hive.options";
058
059    public static final String HIVE_SCRIPT_PARAMS = "oozie.hive.script.params";
060
061    public static final String SQOOP_COMMAND = "oozie.sqoop.command";
062
063    public static final String SQOOP_OPTIONS = "oozie.sqoop.options";
064
065    public static final String FILES = "oozie.files";
066
067    public static final String ARCHIVES = "oozie.archives";
068
069    public static final String IS_PROXY_SUBMISSION = "oozie.proxysubmission";
070
071    protected XOozieClient() {
072    }
073
074    /**
075     * Create an eXtended Workflow client instance.
076     *
077     * @param oozieUrl URL of the Oozie instance it will interact with.
078     */
079    public XOozieClient(String oozieUrl) {
080        super(oozieUrl);
081    }
082
083    private String readScript(String script) throws IOException {
084        if (!new File(script).exists()) {
085            throw new IOException("Error: script file [" + script + "] does not exist");
086        }
087
088        BufferedReader br = null;
089        try {
090            br = new BufferedReader(new FileReader(script));
091            StringBuilder sb = new StringBuilder();
092            String line;
093            while ((line = br.readLine()) != null) {
094                sb.append(line + "\n");
095            }
096            return sb.toString();
097        }
098        finally {
099            try {
100                br.close();
101            }
102            catch (IOException ex) {
103                System.err.println("Error: " + ex.getMessage());
104            }
105        }
106    }
107
108    private String serializeSqoopCommand(String[] command) {
109        StringBuilder sb = new StringBuilder();
110        for (String arg : command) {
111            sb.append(arg).append("\n");
112        }
113        return sb.toString();
114    }
115
116    static void setStrings(Properties conf, String key, String[] values) {
117        if (values != null) {
118            conf.setProperty(key + ".size", (new Integer(values.length)).toString());
119            for (int i = 0; i < values.length; i++) {
120                conf.setProperty(key + "." + i, values[i]);
121            }
122        }
123    }
124
125    private void validateHttpSubmitConf(Properties conf) {
126        String JT = conf.getProperty(XOozieClient.JT);
127        String JT_2 = conf.getProperty(XOozieClient.JT_2);
128        if (JT == null) {
129            if(JT_2 == null) {
130                throw new RuntimeException("jobtracker is not specified in conf");
131            }
132        }
133
134        String NN = conf.getProperty(XOozieClient.NN);
135        String NN_2 = conf.getProperty(XOozieClient.NN_2);
136        if (NN == null) {
137            if(NN_2 == null) {
138                throw new RuntimeException("namenode is not specified in conf");
139            } else {
140                NN = NN_2;
141            }
142        }
143
144        String libPath = conf.getProperty(LIBPATH);
145        if (libPath == null) {
146            throw new RuntimeException("libpath is not specified in conf");
147        }
148        if (!libPath.contains(":/")) {
149            String newLibPath;
150            if (libPath.startsWith("/")) {
151                if(NN.endsWith("/")) {
152                    newLibPath = NN + libPath.substring(1);
153                } else {
154                    newLibPath = NN + libPath;
155                }
156            } else {
157                throw new RuntimeException("libpath should be absolute");
158            }
159            conf.setProperty(LIBPATH, newLibPath);
160        }
161
162        conf.setProperty(IS_PROXY_SUBMISSION, "true");
163    }
164
165    /**
166     * Submit a Pig job via HTTP.
167     *
168     * @param conf job configuration.
169     * @param pigScriptFile pig script file.
170     * @param pigArgs pig arguments string.
171     * @return the job Id.
172     * @throws OozieClientException thrown if the job could not be submitted.
173     */
174    @Deprecated
175    public String submitPig(Properties conf, String pigScriptFile, String[] pigArgs) throws IOException, OozieClientException {
176        return submitScriptLanguage(conf, pigScriptFile, pigArgs, OozieCLI.PIG_CMD);
177    }
178
179    /**
180     * Submit a Pig or Hive job via HTTP.
181     *
182     * @param conf job configuration.
183     * @param scriptFile  script file.
184     * @param args  arguments string.
185     * @return the job Id.
186     * @throws OozieClientException thrown if the job could not be submitted.
187     */
188    public String submitScriptLanguage(Properties conf, String scriptFile, String[] args, String jobType)
189            throws IOException, OozieClientException {
190        return submitScriptLanguage(conf, scriptFile, args, null, jobType);
191    }
192
193    /**
194     * Submit a Pig or Hive job via HTTP.
195     *
196     * @param conf job configuration.
197     * @param scriptFile  script file.
198     * @param args  arguments string.
199     * @param params parameters string.
200     * @return the job Id.
201     * @throws OozieClientException thrown if the job could not be submitted.
202     */
203    public String submitScriptLanguage(Properties conf, String scriptFile, String[] args, String[] params, String jobType)
204            throws IOException, OozieClientException {
205        if (conf == null) {
206            throw new IllegalArgumentException("conf cannot be null");
207        }
208        if (scriptFile == null) {
209            throw new IllegalArgumentException("scriptFile cannot be null");
210        }
211
212        validateHttpSubmitConf(conf);
213
214        String script = "";
215        String options = "";
216        String scriptParams = "";
217
218        if (jobType.equals(OozieCLI.HIVE_CMD)) {
219            script = XOozieClient.HIVE_SCRIPT;
220            options = XOozieClient.HIVE_OPTIONS;
221            scriptParams = XOozieClient.HIVE_SCRIPT_PARAMS;
222        }
223        else if (jobType.equals(OozieCLI.PIG_CMD)) {
224            script =  XOozieClient.PIG_SCRIPT;
225            options = XOozieClient.PIG_OPTIONS;
226            scriptParams = XOozieClient.PIG_SCRIPT_PARAMS;
227        }
228        else {
229            throw new IllegalArgumentException("jobType must be either pig or hive");
230        }
231
232        conf.setProperty(script, readScript(scriptFile));
233        setStrings(conf, options, args);
234        setStrings(conf, scriptParams, params);
235
236        return (new HttpJobSubmit(conf, jobType)).call();
237    }
238
239    /**
240     * Submit a Sqoop job via HTTP.
241     *
242     * @param conf job configuration.
243     * @param command sqoop command to run.
244     * @param args  arguments string.
245     * @return the job Id.
246     * @throws OozieClientException thrown if the job could not be submitted.
247     */
248    public String submitSqoop(Properties conf, String[] command, String[] args)
249            throws OozieClientException {
250        if (conf == null) {
251            throw new IllegalArgumentException("conf cannot be null");
252        }
253        if (command == null) {
254            throw new IllegalArgumentException("command cannot be null");
255        }
256
257        validateHttpSubmitConf(conf);
258
259        conf.setProperty(XOozieClient.SQOOP_COMMAND, serializeSqoopCommand(command));
260        setStrings(conf, XOozieClient.SQOOP_OPTIONS, args);
261
262        return (new HttpJobSubmit(conf, OozieCLI.SQOOP_CMD)).call();
263    }
264
265    /**
266     * Submit a Map/Reduce job via HTTP.
267     *
268     * @param conf job configuration.
269     * @return the job Id.
270     * @throws OozieClientException thrown if the job could not be submitted.
271     */
272    public String submitMapReduce(Properties conf) throws OozieClientException {
273        if (conf == null) {
274            throw new IllegalArgumentException("conf cannot be null");
275        }
276
277        validateHttpSubmitConf(conf);
278
279        return (new HttpJobSubmit(conf, "mapreduce")).call();
280    }
281
282    private class HttpJobSubmit extends ClientCallable<String> {
283        private Properties conf;
284
285        HttpJobSubmit(Properties conf, String jobType) {
286            super("POST", RestConstants.JOBS, "", prepareParams(RestConstants.JOBTYPE_PARAM, jobType));
287            this.conf = notNull(conf, "conf");
288        }
289
290        @Override
291        protected String call(HttpURLConnection conn) throws IOException, OozieClientException {
292            conn.setRequestProperty("content-type", RestConstants.XML_CONTENT_TYPE);
293            writeToXml(conf, conn.getOutputStream());
294            if (conn.getResponseCode() == HttpURLConnection.HTTP_CREATED) {
295                JSONObject json = (JSONObject) JSONValue.parse(new InputStreamReader(conn.getInputStream()));
296                return (String) json.get(JsonTags.JOB_ID);
297            }
298            if (conn.getResponseCode() != HttpURLConnection.HTTP_OK) {
299                handleError(conn);
300            }
301            return null;
302        }
303    }
304
305    /**
306     * set LIBPATH for HTTP submission job.
307     *
308     * @param conf Configuration object.
309     * @param pathStr lib HDFS path.
310     */
311    public void setLib(Properties conf, String pathStr) {
312        conf.setProperty(LIBPATH, pathStr);
313    }
314
315    /**
316     * The equivalent to <file> tag in oozie's workflow xml.
317     *
318     * @param conf Configuration object.
319     * @param file file HDFS path. A "#..." symbolic string can be appended to the path to specify symbolic link name.
320     *             For example, "/user/oozie/parameter_file#myparams". If no "#..." is specified, file name will be used as
321     *             symbolic link name.
322     */
323    public void addFile(Properties conf, String file) {
324        if (file == null || file.length() == 0) {
325            throw new IllegalArgumentException("file cannot be null or empty");
326        }
327        String files = conf.getProperty(FILES);
328        conf.setProperty(FILES, files == null ? file : files + "," + file);
329    }
330
331    /**
332     * The equivalent to <archive> tag in oozie's workflow xml.
333     *
334     * @param conf Configuration object.
335     * @param file file HDFS path. A "#..." symbolic string can be appended to the path to specify symbolic link name.
336     *             For example, "/user/oozie/udf1.jar#my.jar". If no "#..." is specified, file name will be used as
337     *             symbolic link name.
338     */
339    public void addArchive(Properties conf, String file) {
340        if (file == null || file.length() == 0) {
341            throw new IllegalArgumentException("file cannot be null or empty");
342        }
343        String files = conf.getProperty(ARCHIVES);
344        conf.setProperty(ARCHIVES, files == null ? file : files + "," + file);
345    }
346}