oozie工作流相关入门整理

 
 
Oozie支持工作流,其定义通过将多个Hadoop Job的定义按照一定的顺序组织起来,然后作为一个整体按照既定的路径运行。一个工作流已经定义了,通过启动该工作流Job,就会执行该工作流中包含的多个Hadoop Job,直到完成,这就是工作流Job的生命周期。
 
通过最简单的一个例子来了解oozie是如何运行的,运行oozie的服务器必须能够访问HDFS,可以提交hadoop mapreduce任务(如果需要执行hive,spark等任务,同样需要对应的环境)。
 
为了让实例比较简单,任务只是在提交后执行一次,并没有使用coordinator,只是使用了workflow.xml,在workflow中定义任务的整体流程,workflow.xml的定义内容简要如下:
<workflow-app xmlns="uri:oozie:workflow:0.4" name="${sys_name}-MonitorPlus-Task-${task_id}">
<start to=“check-xxx-succ-flag"/> <decision name=“check-xxx-succ-flag">
<switch>
<case to="check-mr-succ-flag">${fs:exists(concat(concat("/xxx/output/xxxList/",
task_id),"/_SUCCESS"))}</case>
<default to=“do-xxx"/>
</switch>
</decision> <decision name="check-mr-succ-flag">
<switch>
<case to="end">${fs:exists(concat(concat(“/xxx/output/", task_id),"/_SUCCESS"))}</case>
<default to="do-mr"/>
</switch>
</decision> <action name=“do-xxx" retry-max="3" retry-interval="10">
<java>
<job-tracker>${jobTracker}</job-tracker>
<name-node>${namenode}</name-node>
<configuration>
<property>
<name>mapreduce.job.queuename</name>
<value>${queueName}</value>
</property>
</configuration>
<main-class>com.xxx.Main</main-class>
...
 
类似于工作流,其中有开始节点 start,一些decision用于流程判断分支节点,以及具体的action,action中可以选择java, shell等类型节点,启动MapReduce任务或者可以选择跳转到某台服务器去启动shell脚本(理论上,如果可以启动shell脚本,就可以做任何在单机上可以做的事情);分支节点可以根据一些条件,比如某个输出目录的_SUCCESS文件是否存在,将以前执行的某些阶段跳过,在执行action时可以设置重试次数以及重试间隔,以避免某次集群环境问题引起的失败。
 
工作流中可以根据需要设置特定的变量参数,以${varibleName}来定义,这样,使用统一的oozie定义来根据传递参数的不同来执行不同的任务。
 
工作流xml文件需要被放在HDFS上才能被oozie调度,如果在启动需要调度MapReduce任务,同样jar包也需要上传到hdfs上,形成下面的目录结构:
 
/xxx/oozie/workflow.xml
/xxx/oozie/lib
/xxx/oozie/lib/mr-1.7-SNAPSHOT-package.jar
/xxx/oozie/lib/xxx.mr-1.7-SNAPSHOT-package.jar
 
在workflow工作流定义的同级目录下,需要有一个lib目录,在lib目录中存在java节点MapReduce使用的jar包。需要注意的是,oozie并不是使用指定jar包的名称来启动任务的,在lib包中绝对不能存在某个jar包的不同版本,不能够出现多个相同主类。
 
工作流中设置的变量,需要通过一个叫做job.properties的配置文件来填充(不能改名称,如果需要多个,请放到不同的文件夹下),job.properties需要配置下面的参数:
 
namenode=hdfs://ns1
jobTracker=xxx.cn:8032
oozie_url=http://xxx.cn:11000/oozie
oozie.wf.application.path=/xxx/oozie/workflow.xml
 
必须要求设置这三个参数namenode,jobTracker,oozie_url,以及任务workflow.xml在hdfs上的位置。其他的参数,视具体的workflow.xml而定,启动oozie的方式为:
 
oozie job -oozie ${oozie_url} -config ${job_cfg} -run
如果oozie是在本机运行,不需要-oozie ${oozie_url},任务提交成功后,就会返回一个oozie任务id,可以在oozie web url中查看该任务的具体执行情况。
 
以上是通过命令行来调用oozie,当然也可以通过java oozie client来调用oozie执行,需要依赖pom:
<dependency>
<groupId>org.apache.oozie</groupId>
<artifactId>oozie-client</artifactId>
<version>4.1.0</version>
</dependency>
可以参考oozie官网中介绍的例子来完成任务提交,这里参照之前的例子,用java实现了一遍:
 
public class WorkflowClient {

    public static final String OOZIE_URL = "http://xxx.cn:11000/oozie";

    private OozieClient oozieClient = new OozieClient(OOZIE_URL);

    public WorkflowClient() {

    }

    /**
* @param workflowDefinition
* @param workflowParameters
* @return - jobId
* @throws org.apache.oozie.client.OozieClientException
*/
public String startJob(Properties properties)
throws OozieClientException {
// create a workflow job configuration and set the workflow application path
Properties configuration = oozieClient.createConfiguration(); Enumeration<?> enumeration = properties.propertyNames();
while (enumeration.hasMoreElements()) {
Object element = enumeration.nextElement();
configuration.setProperty(element.toString(), properties.getProperty(element.toString()).toString());
} return oozieClient.run(configuration);
} public static void main(String[] args) throws OozieClientException, InterruptedException, IOException {
String jobPropertyFile = args[0]; // Create client
WorkflowClient client = new WorkflowClient();
// Create parameters
Properties properties = new Properties();
properties.load(new FileInputStream(new File(jobPropertyFile))); // Start Oozing
String jobId = client.startJob(properties);
SysOutLogger.info("jobId: " + jobId);
WorkflowJob jobInfo = client.oozieClient.getJobInfo(jobId); SysOutLogger.info("job url: " + jobInfo.getConsoleUrl()); while (true) {
Thread.sleep(1000L);
WorkflowJob.Status status = client.oozieClient.getJobInfo(jobId).getStatus();
if (status == WorkflowJob.Status.SUCCEEDED || status == WorkflowJob.Status.FAILED
|| status == WorkflowJob.Status.KILLED) {
SysOutLogger.info("Job finish with status: " + status);
break;
} else {
SysOutLogger.info("job running: " + jobInfo.getStatus());
}
} }
}
 
用maven打包完成后,将其放到服务器端运行:
 
[INFO]  2015-04-16 11:02:18 : jobId: 0000002-150415180413953-oozie-supe-W
[INFO] 2015-04-16 11:02:18 : job url: http://xxx.cn:11000/oozie?job=0000002-150415180413953-oozie-supe-W
[INFO] 2015-04-16 11:02:18 : job running: RUNNING
[INFO] 2015-04-16 11:02:19 : Job finish with status: SUCCEEDED
 
在job提交后,根据jobId,就通过oozieClient获得其WorkflowJob,使用OozieClient对应的API还可以对已经执行的任务进行kill/rerun等操作。由于没有找到ooze对应的事件API,这里是通过定时获取状态的方式将状态打印出来,直到任务完成。从oozie文档中可以查找到,oozie是可以和JMS服务集成,将任务执行状态发送到对应topic上,这样就可以依赖JMS Listener实现类似的事件监听/通知机制。
 
 
上一篇:如何为非常不确定的行为(如并发)设计安全的 API,使用这些 API 时如何确保安全


下一篇:漂亮的Android加载中动画:AVLoadingIndicatorView