图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Hadoop|YARN › 技术应用› Hadoop 的 Oozie 工作流管理引擎的实际应用
分享

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

lovebyz 2015-6-10 20:45:09 发表于 技术应用 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 0 23807
                       
    1. Oozie shell 操作 ! ?
    2. Oozie Java 操作  ! ?
    3. Oozie Sqoop 操作 ! ?
    4. Oozie Hive 操作  ! ?
    5. Oozie 电子邮件操作 ! ?



Apache Oozie 是 Apache Hadoop 生态系统中的一个关键组件，它使得开发人员能够调度电子邮件通知方面的重复作业，或者调度使用Java™、UNIX Shell、Apache Hive、Apache Pig 和 Apache Sqoop等各种编程语言编写的重复作业。本文将介绍处理某些业务场景的框架的实际应用，还将谈论解决方案架构师、技术架构师、顾问、数据科学家、技术领导，以及致力于大数据技术或对大数据技术感兴趣的开发人员等方面的话题。
               




简介
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 编排架构
tu1 (2).png


需要安装的软件
要想充分利用本文的示例，访问以下软件可能对您有所帮助：
该集群是一个分布式集群，在 1 个主名称节点、2 个核心节点和 8 个任务节点中运行。

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

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

清单 1. 启动控制节点[mw_shl_code=html,true]<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf"> <start to="timeCheck"/> </workflow-app>[/mw_shl_code]
末端控制节点（如清单 2 所示）是结束工作流作业时所用的节点。它表示工作流操作已经成功完成。一个工作流定义必须有一个末端节点。

清单 2. 末端控制节点[mw_shl_code=html,false]<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf">
    <end name="end"/>
</workflow-app>[/mw_shl_code]
停止控制节点（如清单 3 所示）可使工作流作业自行停止。在到达停止节点（kill                node）时，如果工作流作业启动的一个或更多操作正在运行，那么当前运行的所有操作都将停止。工作流定义可以包含零个或更多停止节点。

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

清单 4. 决策控制节点[mw_shl_code=html,false]<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>[/mw_shl_code]
分支节点 将一个执行路径分为多个并发路径。联接节点                一直等待，直到前面的分支节点的所有并发执行路径都到达联接节点为止。您必须成对使用分叉节点和联接节点，如清单 5 所示。

清单 5. 分支-联接控制节点[mw_shl_code=html,false]<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>[/mw_shl_code]


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 脚本[mw_shl_code=bash,false]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[/mw_shl_code]"

清单 7 展示了 workflow.xml 文件中的 shell 操作配置。
清单 7. Oozie shell 操作[mw_shl_code=bash,false]<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>[/mw_shl_code]
要想访问 shell 输出，可以使用清单 8 中所示的 Sqoop 增量作业。

清单 8. 用来实现增量导入的 Oozie Sqoop 操作[mw_shl_code=bash,false]<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>[/mw_shl_code]


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 操作[mw_shl_code=java,false]<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>[/mw_shl_code]
您可以通过使用 capture-output 来配置 Java 操作，从而将值传递给下一操作。您可以通过使用 Hadoop EL                函数来访问这些值。您能够以 Java 属性文件的格式在 Java 类中写入值，如清单 10 所示。

清单 10. 用于实现值传递的 Java 代码片段[mw_shl_code=java,false]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");
    }[/mw_shl_code]
您可以在 workflow.xml 文件中配置操作来访问属性文件中的相应值设置，如清单 11 所示。

清单 11. 用于实现值传递的 Oozie Java 操作[mw_shl_code=java,false]<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>[/mw_shl_code]


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 操作[mw_shl_code=html,false]<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>[/mw_shl_code]


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 操作[mw_shl_code=html,false]<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>[/mw_shl_code]


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

清单 14. Oozie 电子邮件操作[mw_shl_code=html,false]<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>[/mw_shl_code]


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




回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
lovebyz

中级会员

关注

  • 9

    主题

  • 150

    帖子

  • 7

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2025 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000