控制节点执行器:
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如何执行并且将结果反馈的;
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);