Oozie-ActionExecutor

控制节点执行器：

StartActionExecutor.class
EndActionExecutor.class
KillActionExecutor.class
ForkActionExecutor.class
JoinActionExecutor.class

任务执行器：

org.apache.oozie.action.decision.DecisionActionExecutor
org.apache.oozie.action.hadoop.JavaActionExecutor
org.apache.oozie.action.hadoop.FsActionExecutor
org.apache.oozie.action.hadoop.MapReduceActionExecutor
org.apache.oozie.action.hadoop.PigActionExecutor
org.apache.oozie.action.hadoop.HiveActionExecutor
org.apache.oozie.action.hadoop.ShellActionExecutor
org.apache.oozie.action.hadoop.SqoopActionExecutor
org.apache.oozie.action.hadoop.DistcpActionExecutor
org.apache.oozie.action.hadoop.Hive2ActionExecutor
org.apache.oozie.action.ssh.SshActionExecutor
org.apache.oozie.action.oozie.SubWorkflowActionExecutor
org.apache.oozie.action.email.EmailActionExecutor
org.apache.oozie.action.hadoop.SparkActionExecutor

本文关注点是wf满足了条件之后 action如何执行并且将结果反馈的；

diagram4.png

Oozie系统没有自己专属的执行机器，oozie采用的策略是 将不同的任务都封装成一个 map-reduce任务，提交到hadoop集群来执行，mr任务结束，任务即结束

public class LauncherMapper<K1, V1, K2, V2> implements Mapper<K1, V1, K2, V2>, Runnable {

如果大家写过map-reduce 任务的话，应该可以想到它的实现逻辑。

public void map(K1 key, V1 value, OutputCollector<K2, V2> collector, Reporter reporter) throws IOException {
    try {
        if (configFailure) {
            throw configureFailureEx;
        }
        else {
            String mainClass = getJobConf().get(CONF_OOZIE_ACTION_MAIN_CLASS);
            if (getJobConf().getBoolean("oozie.hadoop-2.0.2-alpha.workaround.for.distributed.cache", false)) {              System.err.println("WARNING, workaround for Hadoop 2.0.2-alpha distributed cached issue (MAPREDUCE-4820) enabled");
            }
            String msgPrefix = "Main class [" + mainClass + "], ";
            int errorCode = 0;
            Throwable errorCause = null;
            String errorMessage = null;
            try {
                new LauncherSecurityManager();
            }
            catch (SecurityException ex) {
                errorMessage = "Could not set LauncherSecurityManager";
                errorCause = ex;
            }
            try {
                setupHeartBeater(reporter);
                setupMainConfiguration();
                // Propagating the conf to use by child job.
                propagateToHadoopConf();
                try {
                    System.out.println("Starting the execution of prepare actions");
                    executePrepare();
                    System.out.println("Completed the execution of prepare actions successfully");
                } catch (Exception ex) {
                    System.out.println("Prepare execution in the Launcher Mapper has failed");
                    throw new LauncherException(ex.getMessage(), ex);
                }
                String[] args = getMainArguments(getJobConf());
                printContentsOfCurrentDir();
                System.out.println();
                System.out.println("Oozie Java/Map-Reduce/Pig action launcher-job configuration");
                System.out.println("=================================================================");
                System.out.println("Workflow job id   : " + System.getProperty("oozie.job.id"));
                System.out.println("Workflow action id: " + System.getProperty("oozie.action.id"));
                System.out.println();
                System.out.println("Classpath         :");
                System.out.println("------------------------");
                StringTokenizer st = new StringTokenizer(System.getProperty("java.class.path"), ":");
                while (st.hasMoreTokens()) {
                    System.out.println("  " + st.nextToken());
                }
                System.out.println("------------------------");
                System.out.println();
                System.out.println("Main class        : " + mainClass);
                System.out.println();
                System.out.println("Maximum output    : "
                        + getJobConf().getInt(CONF_OOZIE_ACTION_MAX_OUTPUT_DATA, 2 * 1024));
                System.out.println();
                System.out.println("Arguments         :");
                for (String arg : args) {
                    System.out.println("                    " + arg);
                }
                System.out.println();
                System.out.println("Java System Properties:");
                System.out.println("------------------------");
                System.getProperties().store(System.out, "");
                System.out.flush();
                System.out.println("------------------------");
                System.out.println();
                System.out.println("=================================================================");

public static void setupLauncherInfo(JobConf launcherConf, String jobId, String actionId, Path actionDir,            String recoveryId, Configuration actionConf, String prepareXML) throws IOException, HadoopAccessorException {
        launcherConf.setMapperClass(LauncherMapper.class);
        launcherConf.setSpeculativeExecution(false);
        launcherConf.setNumMapTasks(1);
        launcherConf.setNumReduceTasks(0);
        launcherConf.set(LauncherMapper.OOZIE_JOB_ID, jobId);
        launcherConf.set(LauncherMapper.OOZIE_ACTION_ID, actionId);
        launcherConf.set(LauncherMapper.OOZIE_ACTION_DIR_PATH,
 actionDir.toString());
        launcherConf.set(LauncherMapper.OOZIE_ACTION_RECOVERY_ID, recoveryId);
        launcherConf.set(LauncherMapper.ACTION_PREPARE_XML, prepareXML);
        actionConf.set(LauncherMapper.OOZIE_JOB_ID, jobId);
        actionConf.set(LauncherMapper.OOZIE_ACTION_ID, actionId);