Oozie工作流程定义详解

  1. 云栖社区>
  2. 博客>
  3. 正文

Oozie工作流程定义详解

shiyanjuncn 2016-04-13 16:40:53 浏览1287
展开阅读全文

Oozie工作流程定义是一个DAG(Directed Acyclical Graphs)图,它由控制流节点(Control Flow Nodes)或动作节点(Action Nodes)组成,各个节点又是通过表征转移的箭线(transitions
arrows)互相连通。对于工作流一般对应存在流程定义语言,例如jBPM是jPDL,大多数都是基于XML定义的,Oozie流程定义语言也是基于XML定义的,称为hPDL(Hadoop Process Definition Language)。
下面,我们详细说明工作流定义相关的内容:

工作流生命周期

在Oozie中,工作流的状态可能存在如下几种:

状态 含义说明
PREP 一个工作流Job第一次创建将处于PREP状态,表示工作流Job已经定义,但是没有运行。
RUNNING 当一个已经被创建的工作流Job开始执行的时候,就处于RUNNING状态。它不会达到结束状态,只能因为出错而结束,或者被挂起。
SUSPENDED 一个RUNNING状态的工作流Job会变成SUSPENDED状态,而且它会一直处于该状态,除非这个工作流Job被重新开始执行或者被杀死。
SUCCEEDED 当一个RUNNING状态的工作流Job到达了end节点,它就变成了SUCCEEDED最终完成状态。
KILLED 当一个工作流Job处于被创建后的状态,或者处于RUNNING、SUSPENDED状态时,被杀死,则工作流Job的状态变为KILLED状态。
FAILED 当一个工作流Job不可预期的错误失败而终止,就会变成FAILED状态。

上述各种状态存在相应的转移(工作流程因为某些事件,可能从一个状态跳转到另一个状态),其中合法的状态转移有如下几种,如下表所示:

转移前状态 转移后状态集合
未启动 PREP
PREP RUNNING、KILLED
RUNNING SUSPENDED、SUCCEEDED、KILLED、FAILED
SUSPENDED RUNNING、KILLED

明确上述给出的状态转移空间以后,可以根据实际需要更加灵活地来控制工作流Job的运行。

控制流节点(Control Flow Nodes)

工作流程定义中,控制工作流的开始和结束,以及工作流Job的执行路径的节点,它定义了流程的开始(start节点)和结束(end节点或kill节点),同时提供了一种控制流程执行路径的机制(decision决策节点、fork分支节点、join会签节点)。通过上面提到的各种节点,我们大概应该能够知道它们在工作流中起着怎样的作用。下面,我们看一下不同节点的语法格式:

  • start节点
1 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
2 ...
3 <start to="[NODE-NAME]" />
4 ...
5 </workflow-app>

上面start元素的to属性,指向第一个将要执行的工作流节点。

  • end节点
1 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
2 ...
3 <end name="[NODE-NAME]" />
4 ...
5 </workflow-app>

达到该节点,工作流Job会变成SUCCEEDED状态,表示成功完成。需要注意的是,一个工作流定义必须只能有一个end节点。

  • kill节点
1 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
2 ...
3 <kill name="[NODE-NAME]">
4 <message>[MESSAGE-TO-LOG]</message>
5 </kill>
6 ...
7 </workflow-app>

kill元素的name属性,是要杀死的工作流节点的名称,message元素指定了工作流节点被杀死的备注信息。达到该节点,工作流Job会变成状态KILLED。

  • decision节点
01 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
02 ...
03 <decision name="[NODE-NAME]">
04 <switch>
05 <case to="[NODE_NAME]">[PREDICATE]</case>
06 ...
07 <case to="[NODE_NAME]">[PREDICATE]</case>
08 <default to="[NODE_NAME]" />
09 </switch>
10 </decision>
11 ...
12 </workflow-app>

decision节点通过预定义一组条件,当工作流Job执行到该节点时,会根据其中的条件进行判断选择,满足条件的路径将被执行。decision节点通过switch…case语法来进行路径选择,只要有满足条件的判断,就会执行对应的路径,如果没有可以配置default元素指向的节点。

  • fork节点和join节点
01 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
02 ...
03 <fork name="[FORK-NODE-NAME]">
04 <path start="[NODE-NAME]" />
05 ...
06 <path start="[NODE-NAME]" />
07 </fork>
08 ...
09 <join name="[JOIN-NODE-NAME]" to="[NODE-NAME]" />
10 ...
11 </workflow-app>

for元素下面会有多个path元素,指定了可以并发执行的多个执行路径。fork中多个并发执行路径会在join节点的位置会合,只有所有的路径都到达后,才会继续执行join节点。

动作节点(Action Nodes)

工作流程定义中,能够触发一个计算任务(Computation Task)或者处理任务(Processing Task)执行的节点。所有的动作(
Action)都有一些基本的特性,我先首先来看一下:

  • 远程执行

对Oozie来说,动作节点的执行都是远程的,因为Oozie可能部署在一个单独的服务器上,而工作流Job是在Hadoop集群的节点上执行的。即使Oozie在Hadoop集群的某个节点上,它也是处于与Hadoop进行独立无关的JVM示例之中(Oozie部署在Servlet容器当中)。

  • 异步性

动作节点的执行,对于Oozie来说是异步的。Oozie启动一个工作流Job,这个工作流Job便开始执行。Oozie可以通过两种方式来探测工作流Job的执行情况:一种是基于回调机制,对每个任务的执行(可以看成是动作节点的执行)都对应一个唯一的URL,如果任务执行结束或者执行失败,会通过回调这个URL通知Oozie已经完成;另一种就是轮询,Oozie不停地去查询任务执行的完成状态,如果由于网络故障回调机制失败,也会使用轮询的方式来处理。

  • 执行结果要么成功,要么失败

如果动作节点执行成功,则会转向ok节点;如果失败则会转向error节点。

  • 可恢复性

如果一个动作节点执行失败,Oozie提供了一些恢复执行的策略,这个要根据失败的特点来进行:如果是状态转移过程中失败,Oozie会根据指定的重试时间间隔去重新执行;如果不是转移性质的失败,则只能通过手工干预来进行恢复;如果重试恢复执行都没有解决问题,则最终会跳转到error节点。

下面详细介绍Oozie内置支持的动作节点类型,如下所示:

  • Map-Reduce动作

map-reduce动作会在工作流Job中启动一个MapReduce Job任务运行,我们可以详细配置这个MapReduce Job。另外,可以通过map-reduce元素的子元素来配置一些其他的任务,如streaming、pipes、file、archive等等。
下面给出包含这些内容的语法格式说明:

01 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
02 ...
03 <action name="[NODE-NAME]">
04 <map-reduce>
05 <job-tracker>[JOB-TRACKER]</job-tracker>
06 <name-node>[NAME-NODE]</name-node>
07 <prepare>
08 <delete path="[PATH]" />
09 ...
10 <mkdir path="[PATH]" />
11 ...
12 </prepare>
13 <streaming>
14 <mapper>[MAPPER-PROCESS]</mapper>
15 <reducer>[REDUCER-PROCESS]</reducer>
16 <record-reader>[RECORD-READER-CLASS]</record-reader>
17 <record-reader-mapping>[NAME=VALUE]</record-reader-mapping>
18 ...
19 <env>[NAME=VALUE]</env>
20 ...
21 </streaming>
22 <!-- Either streaming or pipes can be specified for an action, not both -->
23 <pipes>
24 <map>[MAPPER]</map>
25 <reduce>
26 [REDUCER]
27 </reducer>
28 <inputformat>[INPUTFORMAT]</inputformat>
29 <partitioner>[PARTITIONER]</partitioner>
30 <writer>[OUTPUTFORMAT]</writer>
31 <program>[EXECUTABLE]</program>
32 </pipes>
33 <job-xml>[JOB-XML-FILE]</job-xml>
34 <configuration>
35 <property>
36 <name>[PROPERTY-NAME]</name>
37 <value>[PROPERTY-VALUE]</value>
38 </property>
39 ...
40 </configuration>
41 <file>[FILE-PATH]</file>
42 ...
43 <archive>[FILE-PATH]</archive>
44 ...
45 </map-reduce>
46 <ok to="[NODE-NAME]" />
47 <error to="[NODE-NAME]" />
48 </action>
49 ...
50 </workflow-app>
  • Hive动作

Hive主要是基于类似SQL的HQL语言的,它能够方便地操作HDFS中数据,实现对海量数据的分析工作。HIve动作的语法格式如下所示:

01 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.2">
02 ...
03 <action name="[NODE-NAME]">
04 <hive xmlns="uri:oozie:hive-action:0.2">
05 <job-tracker>[JOB-TRACKER]</job-tracker>
06 <name-node>[NAME-NODE]</name-node>
07 <prepare>
08 <delete path="[PATH]" />
09 ...
10 <mkdir path="[PATH]" />
11 ...
12 </prepare>
13 <configuration>
14 <property>
15 <name>[PROPERTY-NAME]</name>
16 <value>[PROPERTY-VALUE]</value>
17 </property>
18 ...
19 </configuration>
20 <script>[HIVE-SCRIPT]</script>
21 <param>[PARAM-VALUE]</param>
22 ...
23 </hive>
24 <ok to="[NODE-NAME]" />
25 <error to="[NODE-NAME]" />
26 </action>
27 ...
28 </workflow-app>
  • Sqoop动作

Sqoop是一个能够在Hadoop和结构化存储系统之间进行数据的导入导出的工具,Sqoop动作的语法格式如下:

01 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.2">
02 ...
03 <action name="[NODE-NAME]">
04 <sqoop xmlns="uri:oozie:sqoop-action:0.2">
05 <job-tracker>[JOB-TRACKER]</job-tracker>
06 <name-node>[NAME-NODE]</name-node>
07 <prepare>
08 <delete path="[PATH]" />
09 ...
10 <mkdir path="[PATH]" />
11 ...
12 </prepare>
13 <configuration>
14 <property>
15 <name>[PROPERTY-NAME]</name>
16 <value>[PROPERTY-VALUE]</value>
17 </property>
18 ...
19 </configuration>
20 <command>[SQOOP-COMMAND]</command>
21 <file>[FILE-PATH]</file>
22 ...
23 </sqoop>
24 <ok to="[NODE-NAME]" />
25 <error to="[NODE-NAME]" />
26 </action>
27 ...
28 </workflow-app>
  • Pig动作

pig动作可以启动运行pig脚本实现的Job,在工作流定义中配置的语法格式说明如下:

01 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.2">
02 ...
03 <action name="[NODE-NAME]">
04 <pig>
05 <job-tracker>[JOB-TRACKER]</job-tracker>
06 <name-node>[NAME-NODE]</name-node>
07 <prepare>
08 <delete path="[PATH]" />
09 ...
10 <mkdir path="[PATH]" />
11 ...
12 </prepare>
13 <job-xml>[JOB-XML-FILE]</job-xml>
14 <configuration>
15 <property>
16 <name>[PROPERTY-NAME]</name>
17 <value>[PROPERTY-VALUE]</value>
18 </property>
19 ...
20 </configuration>
21 <script>[PIG-SCRIPT]</script>
22 <param>[PARAM-VALUE]</param>
23 ...
24 <param>[PARAM-VALUE]</param>
25 <argument>[ARGUMENT-VALUE]</argument>
26 ...
27 <argument>[ARGUMENT-VALUE]</argument>
28 <file>[FILE-PATH]</file>
29 ...
30 <archive>[FILE-PATH]</archive>
31 ...
32 </pig>
33 <ok to="[NODE-NAME]" />
34 <error to="[NODE-NAME]" />
35 </action>
36 ...
37 </workflow-app>
  • Fs动作

Fs动作主要是基于HDFS的一些基本操作,如删除路径、创建路径、移动文件、设置文件全乡等等。
语法格式:

01 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
02 ...
03 <action name="[NODE-NAME]">
04 <fs>
05 <delete path='[PATH]' />
06 ...
07 <mkdir path='[PATH]' />
08 ...
09 <move source='[SOURCE-PATH]' target='[TARGET-PATH]' />
10 ...
11 <chmod path='[PATH]' permissions='[PERMISSIONS]' dir-files='false' />
12 ...
13 <touchz path='[PATH]' />
14 </fs>
15 <ok to="[NODE-NAME]" />
16 <error to="[NODE-NAME]" />
17 </action>
18 ...
19 </workflow-app>
  • SSH动作

该动作主要是通过ssh登录到一台主机,能够执行一组shell命令,它在Oozie schema 0.2中已经被删除。
语法格式:

01 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
02 ...
03 <action name="[NODE-NAME]">
04 <ssh>
05 <host>[USER]@[HOST]</host>
06 <command>[SHELL]</command>
07 <args>[ARGUMENTS]</args>
08 ...
09 <capture-output />
10 </ssh>
11 <ok to="[NODE-NAME]" />
12 <error to="[NODE-NAME]" />
13 </action>
14 ...
15 </workflow-app>
  • Java动作

Java动作,是执行一个具有main入口方法的应用程序,在Oozie工作流定义中,会作为一个MapReduce Job执行,这个Job只有一个Map任务。我们需要指定NameNode、JobTracker的信息,还有配置一个Java应用程序的JVM选项参数(java-opts),以及传给主函数(arg)。
语法格式:

01 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
02 ...
03 <action name="[NODE-NAME]">
04 <java>
05 <job-tracker>[JOB-TRACKER]</job-tracker>
06 <name-node>[NAME-NODE]</name-node>
07 <prepare>
08 <delete path="[PATH]" />
09 ...
10 <mkdir path="[PATH]" />
11 ...
12 </prepare>
13 <job-xml>[JOB-XML]</job-xml>
14 <configuration>
15 <property>
16 <name>[PROPERTY-NAME]</name>
17 <value>[PROPERTY-VALUE]</value>
18 </property>
19 ...
20 </configuration>
21 <main-class>[MAIN-CLASS]</main-class>
22 <java-opts>[JAVA-STARTUP-OPTS]</java-opts>
23 <arg>ARGUMENT</arg>
24 ...
25 <file>[FILE-PATH]</file>
26 ...
27 <archive>[FILE-PATH]</archive>
28 ...
29 <capture-output />
30 </java>
31 <ok to="[NODE-NAME]" />
32 <error to="[NODE-NAME]" />
33 </action>
34 ...
35 </workflow-app>
  • Sub-workflow动作

Sub-workflow动作是一个子流程的动作,主流程执行过程中,遇到子流程节点执行时,会一直等待子流程节点执行完成后,才能继续跳转到下一个要执行的节点。
语法格式:

01 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
02 ...
03 <action name="[NODE-NAME]">
04 <sub-workflow>
05 <app-path>[WF-APPLICATION-PATH]</app-path>
06 <propagate-configuration />
07 <configuration>
08 <property>
09 <name>[PROPERTY-NAME]</name>
10 <value>[PROPERTY-VALUE]</value>
11 </property>
12 ...
13 </configuration>
14 </sub-workflow>
15 <ok to="[NODE-NAME]" />
16 <error to="[NODE-NAME]" />
17 </action>
18 ...
19 </workflow-app>
  • Shell动作

Shell动作可以执行Shell命令,并通过配置命令所需要的参数。它的语法格式:

01 <workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.4">
02 ...
03 <action name="[NODE-NAME]">
04 <shell