Oozie workflow nodes (一)

[toc]

Oozie的工作流结点(workflow nodes)分为流控制结点(control flow nodes)和行为结点(action nodes)。

流控制结点：控制工作流的开始、结束以及job的执行路径。

行为结点：触发一个计算或者处理任务执行的节点。

控制流结点

流控制节点定义了工作流的开始和结束（start、end和kill结点），并且提供了一套控制流执行路径的机制（decision，fork和join结点）

Start结点

当工作流启动的时候，它会自动运行在start里面指定的结点。 一个工作流必须有一个start结点。

语法：

<workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
  ...
  <start to="[NODE-NAME]"/>
  ...
</workflow-app>

其中属性to指定执行的结点名。

例子：

<workflow-app name="foo-wf" xmlns="uri:oozie:workflow:0.1">
    ...
    <start to="firstHadoopJob"/>
    ...
</workflow-app>

End结点

当一个工作流任务到达结束结点的时候，该任务会成功结束。当同时有多个任务在运行，而其中一个任务到达end结点。那么其它结点将会被杀死。在这种情况下，工作流任务依然被认为是成功执行的。 一个工作流必须有一个end结点。

语法：

<workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
    ...
    <end name="[NODE-NAME]"/>
    ...
</workflow-app>

属性name的值是结束工作流任务的任务名称。

例子：

<workflow-app name="foo-wf" xmlns="uri:oozie:workflow:0.1">
    ...
    <end name="end"/>
</workflow-app>

Kill结点

kill结点可以让工作流杀掉自己。
当一个job到达kill结点的时候，它会以error（KILLED）的状态退出。
如果有多个行为被启动，而当其中一个到达kill结点的时候，所有的action都会被杀死。
每个工作流有一个或者0个Kill结点。

语法：

<workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
    ...
    <kill name="[NODE-NAME]">
        <message>[MESSAGE-TO-LOG]</message>
    </kill>
    ...
</workflow-app>

属性name是这个kill动作结点的名称。

message的内容将会作为job被kill的原因而打印到日志中。

例子：

<workflow-app name="foo-wf" xmlns="uri:oozie:workflow:0.1">
    ...
    <kill name="killBecauseNoInput">
        <message>Input unavailable</message>
    </kill>
    ...
</workflow-app>

Decision Control结点

决择结点可以让工作流选择执行的路径。

决择结点的行为可以被认为是一个switch-case块。

一个Decision结点包含一组判断-转移对和一个默认转移。按顺序进行判断or appearance until one of them evaluates to true and the corresponding transition is taken.如果没有判断为true，那么将会选择默认的转移。

判断语句是JSP的EL表达式。比如：

${fs:fileSize(‘/usr/foo/myinputdir’) gt 10 * GB}

语法：

<workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
    ...
    <decision name="[NODE-NAME]">
        <switch>
            <case to="[NODE_NAME]">[PREDICATE]</case>
            ...
            <case to="[NODE_NAME]">[PREDICATE]</case>
            <default to="[NODE_NAME]"/>
        </switch>
    </decision>
    ...
</workflow-app>

The name attribute in the decision node is the name of the decision node.

每个case元素包含一个判断语句和一个转移名称。判断语句会被按顺序评估，直到一个返回true，从而获得相应的转移名称。

The default element indicates the transition to take if none of the predicates evaluates to true .

所有的 decision结点必须有一个默认的元素，从而避免当没有判断返回true的时候，工作流进入一个错误的状态。

例子：

<workflow-app name="foo-wf" xmlns="uri:oozie:workflow:0.1">
    ...
    <decision name="mydecision">
        <switch>
            <case to="reconsolidatejob">
              ${fs:fileSize(secondjobOutputDir) gt 10 * GB}
            </case>
            <case to="rexpandjob">
              ${fs:filSize(secondjobOutputDir) lt 100 * MB}
            </case>
            <case to="recomputejob">
              ${ hadoop:counters('secondjob')[RECORDS][REDUCE_OUT] lt 1000000 }
            </case>
            <default to="end"/>
        </switch>
    </decision>
    ...
</workflow-app>

 Fork 和 Join 结点

Fork结点把工作流运行的一条路径变成多个并行执行的路径。

Join结点则等待每个由上一个fork结点产生的并行执行的路径到达它。

Fork和Join结点必须在一起使用。Join结点假定并发执行的路径都是同一个fork结点产生的。

语法：

<workflow-app name="[WF-DEF-NAME]" xmlns="uri:oozie:workflow:0.1">
    ...
    <fork name="[FORK-NODE-NAME]">
        <path start="[NODE-NAME]" />
        ...
        <path start="[NODE-NAME]" />
    </fork>
    ...
    <join name="[JOIN-NODE-NAME]" to="[NODE-NAME]" />
    ...
</workflow-app>

fork结点的name属性表示fork结点的名称。在path元素中的start属性指定要并发执行的结点名称。

join结点的name属性也是它的名称。to属性指定当所有并行运行的路径到达join结点后，下一步运行的结点名称。

例子：

<workflow-app name="sample-wf" xmlns="uri:oozie:workflow:0.1">
    ...
    <fork name="forking">
        <path start="firstparalleljob"/>
        <path start="secondparalleljob"/>
    </fork>
    <action name="firstparallejob">
        <map-reduce>
            <job-tracker>foo:8021</job-tracker>
            <name-node>bar:8020</name-node>
            <job-xml>job1.xml</job-xml>
        </map-reduce>
        <ok to="joining"/>
        <error to="kill"/>
    </action>
    <action name="secondparalleljob">
        <map-reduce>
            <job-tracker>foo:8021</job-tracker>
            <name-node>bar:8020</name-node>
            <job-xml>job2.xml</job-xml>
        </map-reduce>
        <ok to="joining"/>
        <error to="kill"/>
    </action>
    <join name="joining" to="nextaction"/>
    ...
</workflow-app>

http://archive.cloudera.com/cdh/3/oozie/WorkflowFunctionalSpec.html#a3_Workflow_Nodes

• ← Oozie Sqoop Action 配置
• Slope One →