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.action.hadoop;
020
021import java.io.IOException;
022import java.io.StringReader;
023import java.net.URISyntaxException;
024import java.util.ArrayList;
025import java.util.List;
026import java.util.Properties;
027import java.util.StringTokenizer;
028
029import org.apache.hadoop.conf.Configuration;
030import org.apache.hadoop.fs.FileSystem;
031import org.apache.hadoop.fs.Path;
032import org.apache.hadoop.mapred.Counters;
033import org.apache.hadoop.mapred.JobClient;
034import org.apache.hadoop.mapred.JobConf;
035import org.apache.hadoop.mapred.JobID;
036import org.apache.hadoop.mapred.RunningJob;
037import org.apache.oozie.action.ActionExecutorException;
038import org.apache.oozie.client.WorkflowAction;
039import org.apache.oozie.service.HadoopAccessorException;
040import org.apache.oozie.util.XConfiguration;
041import org.apache.oozie.util.XmlUtils;
042import org.apache.oozie.util.XLog;
043import org.jdom.Element;
044import org.jdom.JDOMException;
045import org.jdom.Namespace;
046
047public class SqoopActionExecutor extends JavaActionExecutor {
048
049  public static final String OOZIE_ACTION_EXTERNAL_STATS_WRITE = "oozie.action.external.stats.write";
050  private static final String SQOOP_MAIN_CLASS_NAME = "org.apache.oozie.action.hadoop.SqoopMain";
051  static final String SQOOP_ARGS = "oozie.sqoop.args";
052
053    public SqoopActionExecutor() {
054        super("sqoop");
055    }
056
057    @Override
058    public List<Class> getLauncherClasses() {
059        List<Class> classes = new ArrayList<Class>();
060        try {
061            classes.add(Class.forName(SQOOP_MAIN_CLASS_NAME));
062        }
063        catch (ClassNotFoundException e) {
064            throw new RuntimeException("Class not found", e);
065        }
066        return classes;
067    }
068
069    @Override
070    protected String getLauncherMain(Configuration launcherConf, Element actionXml) {
071        return launcherConf.get(LauncherMapper.CONF_OOZIE_ACTION_MAIN_CLASS, SQOOP_MAIN_CLASS_NAME);
072    }
073
074    @Override
075    @SuppressWarnings("unchecked")
076    Configuration setupActionConf(Configuration actionConf, Context context, Element actionXml, Path appPath)
077            throws ActionExecutorException {
078        super.setupActionConf(actionConf, context, actionXml, appPath);
079        Namespace ns = actionXml.getNamespace();
080
081        try {
082            Element e = actionXml.getChild("configuration", ns);
083            if (e != null) {
084                String strConf = XmlUtils.prettyPrint(e).toString();
085                XConfiguration inlineConf = new XConfiguration(new StringReader(strConf));
086                checkForDisallowedProps(inlineConf, "inline configuration");
087                XConfiguration.copy(inlineConf, actionConf);
088            }
089        } catch (IOException ex) {
090            throw convertException(ex);
091        }
092
093        String[] args;
094        if (actionXml.getChild("command", ns) != null) {
095            String command = actionXml.getChild("command", ns).getTextTrim();
096            StringTokenizer st = new StringTokenizer(command, " ");
097            List<String> l = new ArrayList<String>();
098            while (st.hasMoreTokens()) {
099                l.add(st.nextToken());
100            }
101            args = l.toArray(new String[l.size()]);
102        }
103        else {
104            List<Element> eArgs = (List<Element>) actionXml.getChildren("arg", ns);
105            args = new String[eArgs.size()];
106            for (int i = 0; i < eArgs.size(); i++) {
107                args[i] = eArgs.get(i).getTextTrim();
108            }
109        }
110
111        setSqoopCommand(actionConf, args);
112        return actionConf;
113    }
114
115    private void setSqoopCommand(Configuration conf, String[] args) {
116        MapReduceMain.setStrings(conf, SQOOP_ARGS, args);
117    }
118
119    /**
120     * We will gather counters from all executed action Hadoop jobs (e.g. jobs
121     * that moved data, not the launcher itself) and merge them together. There
122     * will be only one job most of the time. The only exception is
123     * import-all-table option that will execute one job per one exported table.
124     *
125     * @param context Action context
126     * @param action Workflow action
127     * @throws ActionExecutorException
128     */
129    @Override
130    public void end(Context context, WorkflowAction action) throws ActionExecutorException {
131        super.end(context, action);
132        JobClient jobClient = null;
133
134        boolean exception = false;
135        try {
136            if (action.getStatus() == WorkflowAction.Status.OK) {
137                Element actionXml = XmlUtils.parseXml(action.getConf());
138                JobConf jobConf = createBaseHadoopConf(context, actionXml);
139                jobClient = createJobClient(context, jobConf);
140
141                // Cumulative counters for all Sqoop mapreduce jobs
142                Counters counters = null;
143
144                // Sqoop do not have to create mapreduce job each time
145                String externalIds = action.getExternalChildIDs();
146                if (externalIds != null && !externalIds.trim().isEmpty()) {
147                    String []jobIds = externalIds.split(",");
148
149                    for(String jobId : jobIds) {
150                        RunningJob runningJob = jobClient.getJob(JobID.forName(jobId));
151                        if (runningJob == null) {
152                          throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "SQOOP001",
153                            "Unknown hadoop job [{0}] associated with action [{1}].  Failing this action!", action
154                            .getExternalId(), action.getId());
155                        }
156
157                        Counters taskCounters = runningJob.getCounters();
158                        if(taskCounters != null) {
159                            if(counters == null) {
160                              counters = taskCounters;
161                            } else {
162                              counters.incrAllCounters(taskCounters);
163                            }
164                        } else {
165                          XLog.getLog(getClass()).warn("Could not find Hadoop Counters for job: [{0}]", jobId);
166                        }
167                    }
168                }
169
170                if (counters != null) {
171                    ActionStats stats = new MRStats(counters);
172                    String statsJsonString = stats.toJSON();
173                    context.setVar(MapReduceActionExecutor.HADOOP_COUNTERS, statsJsonString);
174
175                    // If action stats write property is set to false by user or
176                    // size of stats is greater than the maximum allowed size,
177                    // do not store the action stats
178                    if (Boolean.parseBoolean(evaluateConfigurationProperty(actionXml,
179                            OOZIE_ACTION_EXTERNAL_STATS_WRITE, "true"))
180                            && (statsJsonString.getBytes().length <= getMaxExternalStatsSize())) {
181                        context.setExecutionStats(statsJsonString);
182                        LOG.debug(
183                          "Printing stats for sqoop action as a JSON string : [{0}]", statsJsonString);
184                    }
185                } else {
186                    context.setVar(MapReduceActionExecutor.HADOOP_COUNTERS, "");
187                    XLog.getLog(getClass()).warn("Can't find any associated Hadoop job counters");
188                }
189            }
190        }
191        catch (Exception ex) {
192            exception = true;
193            throw convertException(ex);
194        }
195        finally {
196            if (jobClient != null) {
197                try {
198                    jobClient.close();
199                }
200                catch (Exception e) {
201                    if (exception) {
202                        LOG.error("JobClient error: ", e);
203                    }
204                    else {
205                        throw convertException(e);
206                    }
207                }
208            }
209        }
210    }
211
212    // Return the value of the specified configuration property
213    private String evaluateConfigurationProperty(Element actionConf, String key, String defaultValue)
214            throws ActionExecutorException {
215        try {
216            if (actionConf != null) {
217                Namespace ns = actionConf.getNamespace();
218                Element e = actionConf.getChild("configuration", ns);
219
220                if(e != null) {
221                  String strConf = XmlUtils.prettyPrint(e).toString();
222                  XConfiguration inlineConf = new XConfiguration(new StringReader(strConf));
223                  return inlineConf.get(key, defaultValue);
224                }
225            }
226            return defaultValue;
227        }
228        catch (IOException ex) {
229            throw convertException(ex);
230        }
231    }
232
233    /**
234     * Get the stats and external child IDs
235     *
236     * @param actionFs the FileSystem object
237     * @param runningJob the runningJob
238     * @param action the Workflow action
239     * @param context executor context
240     *
241     */
242    @Override
243    protected void getActionData(FileSystem actionFs, RunningJob runningJob, WorkflowAction action, Context context)
244            throws HadoopAccessorException, JDOMException, IOException, URISyntaxException{
245        super.getActionData(actionFs, runningJob, action, context);
246        readExternalChildIDs(action, context);
247    }
248
249    @Override
250    protected boolean getCaptureOutput(WorkflowAction action) throws JDOMException {
251        return true;
252    }
253
254
255    /**
256     * Return the sharelib name for the action.
257     *
258     * @return returns <code>sqoop</code>.
259     * @param actionXml
260     */
261    @Override
262    protected String getDefaultShareLibName(Element actionXml) {
263        return "sqoop";
264    }
265
266}