针对 Hadoop 的 Oozie 工作流管理引擎的实际应用

简介

Apache Oozie 是用于 Hadoop 平台的一种工作流调度引擎。该框架（如图 1 所示）使用 Oozie 协调器促进了相互依赖的重复工作之间的协调，您可以使用预定的时间或数据可用性来触发 Apache Oozie。您可以使用 Oozie bundle 系统提交或维护一组协调应用程序。作为本练习的一部分，Oozie 运行了一个 Apache Sqoop 作业，以便在 MySQL 数据库中的数据上执行导入操作，并将数据传输到 Hadoop 分布式文件系统 (HDFS) 中。可以利用导入的数据集执行 Sqoop 合并操作，从而更新较旧的数据集。通过利用 UNIX shell 操作，可从 MySQL 数据库中提取用来执行 Sqoop 作业的元数据。同理，可执行 Java 操作来更新 Sqoop 作业所需的 MySQL 数据库中的元数据。

图 1. Oozie 编排架构

回页首

需要安装的软件

要想充分利用本文的示例，访问以下软件可能对您有所帮助：

• Hadoop 2.2.0
• 一个 MySQL 数据库
• Oozie-4.0.0
• Hive-0.11.0
• 电子邮件服务
• 本文可下载的源代码，便于参考

该集群是一个分布式集群，在 1 个主名称节点、2 个核心节点和 8 个任务节点中运行。

回页首

Oozie 工作流

Oozie 工作流是控制依赖有向非循环图 (DAG) 中安排的 Oozie 操作的集合。控制依赖（Control dependency）可确保以下操作在前面的操作已成功完成后才会启动。本文首先会简要概述工作流控制节点，然后重点介绍以下工作流操作节点：

• Oozie shell 操作
• Oozie Java 操作
• Oozie Sqoop 操作
• Oozie Hive 操作
• Oozie 电子邮件操作

回页首

工作流控制节点

启动控制节点 （如清单 1 所示）是工作流作业的入口点。在工作流启动时，它会自动过渡到启动过程中指定的节点。

清单 1. 启动控制节点

<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf">     <start to="timeCheck"/> </workflow-app>

末端控制节点 （如清单 2 所示）是结束工作流作业时所用的节点。它表示工作流操作已经成功完成。一个工作流定义必须有一个末端节点。

清单 2. 末端控制节点

<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf">     <end name="end"/> </workflow-app>

停止控制节点 （如清单 3 所示）可使工作流作业自行停止。在到达停止节点（kill node）时，如果工作流作业启动的一个或更多操作正在运行，那么当前运行的所有操作都将停止。工作流定义可以包含零个或更多停止节点。

清单 3. 停止控制节点

<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf">     <kill name="fail">         <message>Sqoop failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>    </kill> </workflow-app>

决策控制节点 （如清单 4 所示）可使工作流确定要采用的执行路径。决策节点的工作原理类似于拥有一组谓词转换对（predicates-transition pair）和一个默认转换的 switch-case 块。谓词是按顺序进行评估的，直至其中一个评估为 ture 为止，同时还会进行相应的转换。如果没有一个谓词被评估为 true，则会采用默认转换。

清单 4. 决策控制节点

<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf">     <decision name="master-decision">        <switch>          <case to="sqoopMerge1">                  ${wf:actionData('hiveSwitch')['paramNum'] eq 1}          </case>          <default to="sqoopMerge2"/>        </switch>    </decision> </workflow-app>

分支节点 将一个执行路径分为多个并发路径。 联接节点 一直等待，直到前面的分支节点的所有并发执行路径都到达联接节点为止。您必须成对使用分叉节点和联接节点，如清单 5 所示。

清单 5. 分支-联接控制节点

<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf">     <fork name="forking">         <path start="sqoopMerge1"/>         <path start="sqoopMerge2"/>     </fork>     <join name="joining" to="hiveSwitch"/> </workflow-app>

回页首

Oozie shell 操作

您可以将 Oozie shell 操作作为工作流的一部分进行配置，从而运行某个文件中的一组 shell 脚本。您可以利用包含必要参数的 job-tracker 、 name-node 和 exec 元素来配置 Oozie shell 操作，从而执行相关的任务，如清单 6 所示。您可以将具有配置参数的一个配置 shell 操作来创建或删除 HDFS 上的文件和目录，然后启动 shell 作业。您可以使用与配置元素内联的 job-xml 元素，利用配置参数将一个 XML 文件传递给 shell 作业。您可以配置其他文件或归档文件，让它们可用于 shell 作业。在 shell 作业结束后，您可以让 shell 作业的输出可供 workflow 作业使用，但它需要满足以下条件：

• 输出的格式必须是一个有效的 Java 属性文件。
• 输出的大小必须小于 2KB。

清单 6. Shell 脚本

host="XXX.XX.XX.XXX" port="3306" username="root" password="" database="zzz" tableName="$1"  #################################### echo "Host: $host" echo "Database: $database" echo "Table: $tableName" ####################################  sqoopLstUpd=`mysql --host=$host --port=$port --user=$username --password=$password  -N -e 'SELECT PARM_DATE_VAL from T_CONTROL_PARM where PARM_NM="SQOOP_INCR_LST_UPD"  and PARM_GROUP_NM="'$tableName'"' $database`  echo "sqoopLstUpd=$sqoopLstUpd" echo "tableName=$tableName"

清单 7 展示了 workflow.xml 文件中的 shell 操作配置。

清单 7. Oozie shell 操作

<action name="timeCheck">  <shell xmlns="uri:oozie:shell-action:0.1">     <job-tracker>${jobTracker}</job-tracker>     <name-node>${nameNode}</name-node>     <configuration>      <property>       <name>mapred.job.queue.name</name>       <value>${queueName}</value>      </property>     </configuration>     <exec>${sqoopUpdTrack}</exec>     <argument>${tableName}</argument>     <file>${sqoopUpdTrackPath}#${sqoopUpdTrack}</file>     <capture-output/>  </shell>  <ok to="sqoopIncrImport"/>  <error to="fail"/> </action> 

要想访问 shell 输出，可以使用清单 8 中所示的 Sqoop 增量作业。

清单 8. 用来实现增量导入的 Oozie Sqoop 操作

<action name="sqoopIncrImport">   <sqoop xmlns="uri:oozie:sqoop-action:0.2">      <job-tracker>${jobTracker}</job-tracker>      <name-node>${nameNode}</name-node>      <prepare>       <delete path="${s3BucketLoc}/${tableName}/incr"/>       <mkdir path="${s3BucketLoc}/${tableName}"/>      </prepare>      <configuration>       <property>        <name>mapred.job.queue.name</name>        <value>${queueName}</value>       </property>      </configuration>      <arg>import</arg>      <arg>--connect</arg>      <arg>${dbURL}</arg>      <arg>--driver</arg>      <arg>${mySqlDriver}</arg>      <arg>--username</arg>      <arg>${user}</arg>      <arg>--table</arg>      <arg>${wf:actionData('timeCheck')['tableName']}</arg>      <arg>--target-dir</arg>      <arg>${s3BucketLoc}/${tableName}/incr</arg>      <arg>--check-column</arg>      <arg>LAST_UPD</arg>      <arg>--incremental</arg>      <arg>lastmodified</arg>      <arg>--last-value</arg>      <arg>${wf:actionData('timeCheck')['sqoopLstUpd']}</arg>      <arg>--m</arg>      <arg>1</arg>     </sqoop>     <ok to="sqoopMetaUpdate"/>     <error to="fail"/>    </action> 

回页首

Oozie Java 操作

Java 操作运行了指定的主要 Java 类的 public static void main ( String [] args ) 方法。Java 应用程序作为具有单个 mapper 任务的 MapReduce 作业运行在 Hadoop 集群上。工作流作业一直要等到 Java 操作结束运行之后才能继续执行下一个操作。Java 操作可使用 job-tracker 、 name-node 、Java 主类、JVM 选项和输入参数进行配置，如清单 9 所示。您可以使用 Expression Language (EL) 表达式将参数分配给内联属性值。您必须以 Java 属性文件的格式写入所有输出参数。

您可以配置 Java 操作来清理 HDFS 文件和目录，或者建立 Apache HCatalog 分区，然后再启动 Java 应用程序。这使得 Oozie 能够在出现暂时性或非暂时性故障时重试 Java 操作。

清单 9. Oozie Java 操作

<action name="sqoopMetaUpdate">     <java>      <job-tracker>${jobTracker}</job-tracker>      <name-node>${nameNode}</name-node>      <configuration>        <property>         <name>mapred.job.queue.name</name>         <value>${queueName}</value>        </property>      </configuration>      <main-class>SqoopMetaUtil</main-class>      <java-opts></java-opts>      <arg>${tableName}</arg>      <archive>${mySqlDriverPath}</archive>     </java>     <ok to="hiveSwitch"/>     <error to="fail"/> </action> 

您可以通过使用 capture-output 来配置 Java 操作，从而将值传递给下一操作。您可以通过使用 Hadoop EL 函数来访问这些值。您能够以 Java 属性文件的格式在 Java 类中写入值，如清单 10 所示。

清单 10. 用于实现值传递的 Java 代码片段

String OOZIE_ACTION_OUTPUT_PROPERTIES = "oozie.action.output.properties"; String oozieProp = System.getProperty(OOZIE_ACTION_OUTPUT_PROPERTIES);    OutputStream os = null;    if(oozieProp != null){   File propFile = new File(oozieProp);   Properties p = new Properties();   p.setProperty("name", "Autodesk");   p.setProperty("address", "Sun Rafael");   try {    os = new FileOutputStream(propFile);    p.store(os, "");   } catch (FileNotFoundException e) {    System.err.println("<<< FileNotFoundException >>>"+e.getMessage());   } catch (IOException e) {    System.err.println("<<< IOException >>>"+e.getMessage());   }   finally{    if(os != null)    try {     os.close();    } catch (IOException e) {     System.err.println("<<< IOException >>>"+e.getMessage());    }   }    }    else{     throw new RuntimeException(OOZIE_ACTION_OUTPUT_PROPERTIES     + " System property not defined");     } 

您可以在 workflow.xml 文件中配置操作来访问属性文件中的相应值设置，如清单 11 所示。

清单 11. 用于实现值传递的 Oozie Java 操作

<action name="jProperties">   <java>      <job-tracker>${jobTracker}</job-tracker>      <name-node>${nameNode}</name-node>      <configuration>      <property>       <name>mapred.job.queue.name</name>       <value>${queueName}</value>      </property>      </configuration>      <main-class>PropertyExplorer</main-class>      <java-opts></java-opts>      <capture-output/>   </java>   <ok to="email"/>   <error to="fail"/>  </action>  <action name="email">    <email xmlns="uri:oozie:email-action:0.1">     <to>surajit.paul@autodesk.com</to>     <subject>Oozie workflow finished successfully!</subject>     <body>${wf:actionData('jProperties')['name']} |      ${wf:actionData('jProperties')['address']}</body>    </email>    <ok to="end"/>    <error to="fail"/>  </action> 

回页首

Oozie Sqoop 操作

Oozie 工作流触发了一个 Sqoop 脚本，该脚本在 Hadoop 集群上启动了一个 Sqoop 作业。Sqoop 作业通过在 Hadoop 集群上启动 MapReduce 作业来完成任务。Sqoop 脚本启动的 MapReduce 作业会将数据从 RDBMS 传输到 HDFS。您可以配置一个 Sqoop 操作（如清单 12 所示）来删除 HDFS 上的文件和目录，然后再启动 Sqoop 作业。与其他 Oozie 操作类似，您可以通过使用 job-xml 元素，利用其他的属性来配置 Sqoop 操作。 configuration 元素中指定的属性值将会覆盖 job-xml 元素中指定的属性。可以将其他文件和归档文件提供给 Sqoop 作业。

清单 12. 用于合并的 Oozie Sqoop 操作

<action name="sqoopMerge1">    <sqoop xmlns="uri:oozie:sqoop-action:0.2">     <job-tracker>${jobTracker}</job-tracker>     <name-node>${nameNode}</name-node>     <prepare>      <delete path="${s3BucketLoc}/${tableName}/master1"/>      <mkdir path="${s3BucketLoc}/${tableName}"/>     </prepare>     <configuration>      <property>       <name>mapred.job.queue.name</name>       <value>${queueName}</value>      </property>     </configuration>     <arg>merge</arg>     <arg>--new-data</arg>     <arg>${s3incr}</arg>     <arg>--onto</arg>     <arg>${s3BucketLoc}/${tableName}/master2</arg>     <arg>--target-dir</arg>     <arg>${s3BucketLoc}/${tableName}/master1</arg>     <arg>--jar-file</arg>     <arg>${tableJarLoc}/${tableName}.jar</arg>     <arg>--class-name</arg>     <arg>${tableName}</arg>     <arg>--merge-key</arg>     <arg>ROW_ID</arg>    </sqoop>    <ok to="hive-master1"/>    <error to="fail"/>    </action> 

回页首

Oozie Hive 操作

您可以配置 Hive 操作（如清单 13 所示）来执行 HDFS 上的文件和目录中的任何 Hive 脚本。该操作启动了一个 MapReduce 作业来完成这些任务。您需要在 Oozie 上配置 Hive 操作，使用 Hive 配置文件 hive-default.xml 或 hive-site.xml 作为 job-xml 元素。对于支持 Hive 访问 Oozie 环境而言，这一点是必需的。您可以配置 Hive 操作来创建或删除 HDFS 文件和目录，然后再启动 Hive 作业。 configuration 元素中指定的属性值将会覆盖 job-xml 文件中指定的值。您可以添加其他的文件和归档文件，让它们可用于 Hive 作业。Oozie 执行了由脚本元素中的路径指定的 Hive 脚本。您可以通过 Oozie 工作流，将参数作为输入参数分配给 Hive 脚本。

清单 13. Oozie Hive 操作

<action name="hiveSwitch">   <shell xmlns="uri:oozie:shell-action:0.1">    <job-tracker>${jobTracker}</job-tracker>    <name-node>${nameNode}</name-node>     <configuration>     <property>      <name>mapred.job.queue.name</name>      <value>${queueName}</value>     </property>     </configuration>     <exec>${hiveSwitchScript}</exec>     <argument>${tableName}</argument>     <file>${hiveSwitchScriptPath}#${hiveSwitchScript}</file>     <capture-output/>     </shell>     <ok to="master-decision"/>     <error to="fail"/>    </action> 

回页首

Oozie 电子邮件操作

Oozie 电子邮件操作（如清单 14 所示）可以从工作流应用程序发送电子邮件。除了主题和消息正文之外，电子邮件操作还必须拥有 to 和 cc（可选）地址。您可以使用逗号分割的电子邮件地址向多个收件人发送电子邮件。电子邮件操作可同步运行，而工作流作业一直要等到发送电子邮件之后才会触发下一个操作。您可以使用 Hadoop EL 表达式将参数分配给电子邮件操作。

清单 14. Oozie 电子邮件操作

<action name="email">    <email xmlns="uri:oozie:email-action:0.1">    <to>surajit.paul@autodesk.com</to>    <subject>Oozie workflow finished successfully!</subject>    <body>${wf:actionData('jProperties')['name']} |     ${wf:actionData('jProperties')['address']}</body>         </email>         <ok to="end"/>         <error to="fail"/>    </action> 

回页首

结束语

当多个相互依赖的作业与数据流捆绑在一起的时候，Oozie 工作流就会变成一个数据管道应用程序。Apache Oozie 工作流促进了数据逻辑流程的设计、错误处理、故障转移机制等。您可以配置 Oozie 协调器或捆绑的应用程序来有效地管理工作流，不过，关于这些主题的讨论已超出了本文的讨论范围。一些等效的 Hadoop 工作流引擎包括 Amazon Data Pipeline、Simple Workflow Engine、Azkaban、Cascading 和 Hamake。虽然 Hamake 和 Oozie 是基于 XML 的配置，但 Azkaban 是使用包含键值对的文本文件进行配置的，而 Cascading 是使用 Java API 进行配置的。

回页首

下载