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 org.apache.hadoop.conf.Configuration;
022import org.apache.hadoop.fs.FileSystem;
023import org.apache.hadoop.fs.Path;
024import org.apache.hadoop.mapred.JobConf;
025import org.apache.oozie.action.ActionExecutorException;
026import org.apache.oozie.client.WorkflowAction;
027import org.apache.oozie.service.Services;
028import org.apache.oozie.service.SparkConfigurationService;
029import org.jdom.Element;
030import org.jdom.Namespace;
031
032import java.util.ArrayList;
033import java.util.List;
034import java.util.Map;
035
036public class SparkActionExecutor extends JavaActionExecutor {
037    public static final String SPARK_MAIN_CLASS_NAME = "org.apache.oozie.action.hadoop.SparkMain";
038    public static final String TASK_USER_PRECEDENCE = "mapreduce.task.classpath.user.precedence"; // hadoop-2
039    public static final String TASK_USER_CLASSPATH_PRECEDENCE = "mapreduce.user.classpath.first";  // hadoop-1
040    public static final String SPARK_MASTER = "oozie.spark.master";
041    public static final String SPARK_MODE = "oozie.spark.mode";
042    public static final String SPARK_OPTS = "oozie.spark.spark-opts";
043    public static final String SPARK_JOB_NAME = "oozie.spark.name";
044    public static final String SPARK_CLASS = "oozie.spark.class";
045    public static final String SPARK_JAR = "oozie.spark.jar";
046
047    public SparkActionExecutor() {
048        super("spark");
049    }
050
051    @Override
052    Configuration setupActionConf(Configuration actionConf, Context context, Element actionXml, Path appPath)
053            throws ActionExecutorException {
054        actionConf = super.setupActionConf(actionConf, context, actionXml, appPath);
055        Namespace ns = actionXml.getNamespace();
056
057        String master = actionXml.getChildTextTrim("master", ns);
058        actionConf.set(SPARK_MASTER, master);
059
060        String mode = actionXml.getChildTextTrim("mode", ns);
061        if (mode != null) {
062            actionConf.set(SPARK_MODE, mode);
063        }
064
065        String jobName = actionXml.getChildTextTrim("name", ns);
066        actionConf.set(SPARK_JOB_NAME, jobName);
067
068        String sparkClass = actionXml.getChildTextTrim("class", ns);
069        if (sparkClass != null) {
070            actionConf.set(SPARK_CLASS, sparkClass);
071        }
072
073        String jarLocation = actionXml.getChildTextTrim("jar", ns);
074        actionConf.set(SPARK_JAR, jarLocation);
075
076        StringBuilder sparkOptsSb = new StringBuilder();
077        if (master.startsWith("yarn")) {
078            String resourceManager = actionConf.get(HADOOP_JOB_TRACKER);
079            Map<String, String> sparkConfig = Services.get().get(SparkConfigurationService.class).getSparkConfig(resourceManager);
080            for (Map.Entry<String, String> entry : sparkConfig.entrySet()) {
081                sparkOptsSb.append("--conf ").append(entry.getKey()).append("=").append(entry.getValue()).append(" ");
082            }
083            // CLOUDERA-BUILD: Tell Spark not to localize Hadoop Configs to fix regression on kerberized cluster (CDH-32176)
084            sparkOptsSb.append("--conf spark.yarn.localizeConfig=false ");
085        }
086        String sparkOpts = actionXml.getChildTextTrim("spark-opts", ns);
087        if (sparkOpts != null) {
088            sparkOptsSb.append(sparkOpts);
089        }
090        if (sparkOptsSb.length() > 0) {
091            actionConf.set(SPARK_OPTS, sparkOptsSb.toString().trim());
092        }
093
094        return actionConf;
095    }
096
097    @Override
098    JobConf createLauncherConf(FileSystem actionFs, Context context, WorkflowAction action, Element actionXml,
099                               Configuration actionConf) throws ActionExecutorException {
100
101        JobConf launcherJobConf = super.createLauncherConf(actionFs, context, action, actionXml, actionConf);
102        if (launcherJobConf.get("oozie.launcher." + TASK_USER_PRECEDENCE) == null) {
103            launcherJobConf.set(TASK_USER_PRECEDENCE, "true");
104        }
105        if (launcherJobConf.get("oozie.launcher." + TASK_USER_CLASSPATH_PRECEDENCE) == null) {
106            launcherJobConf.set(TASK_USER_CLASSPATH_PRECEDENCE, "true");
107        }
108        return launcherJobConf;
109    }
110
111    @Override
112    public List<Class> getLauncherClasses() {
113        List<Class> classes = new ArrayList<Class>();
114        try {
115            classes.add(Class.forName(SPARK_MAIN_CLASS_NAME));
116        } catch (ClassNotFoundException e) {
117            throw new RuntimeException("Class not found", e);
118        }
119        return classes;
120    }
121
122
123    /**
124     * Return the sharelib name for the action.
125     *
126     * @param actionXml
127     * @return returns <code>spark</code>.
128     */
129    @Override
130    protected String getDefaultShareLibName(Element actionXml) {
131        return "spark";
132    }
133
134    @Override
135    protected String getLauncherMain(Configuration launcherConf, Element actionXml) {
136        return launcherConf.get(LauncherMapper.CONF_OOZIE_ACTION_MAIN_CLASS, SPARK_MAIN_CLASS_NAME);
137    }
138
139}