Java Web应用程序：Oozie及其使用方式

￡神魔★判官ぃ 2022-03-11 16:24 328阅读 0赞

** 简单示例**

为了向你展示Oozie的使用方法，让我们创建一个简单的示例。我们拥有两个Map/Reduce作业\[3\]——一个会获取最初的数据，另一个会合并指定类型的数据。实际的获取操作需要执行最初的获取操作，然后把两种类型的数据——Lidar和Multicam——合并。为了让这个过程自动化，我们需要创建一个简单的Oozie工作流(代码1)

<workflow-app xmlns='uri:oozie:workflow:0.1' name='NGMB-IPS-ingestion'>  
    <start  to='ingestor'/>  
    <action name='ingestor'>  
        <java>  
            <job-tracker>$\{jobTracker\}</job-tracker>  
            <name-node>$\{nameNode\}</name-node>  
            <configuration>  
                <property>  
                    <name>mapred.job.queue.name</name>  
                    <value>default</value>  
                </property>  
            </configuration>  
            <main-class>com.navteq.assetmgmt.MapReduce.ips.IPSLoader</main-class>  
            <java-opts>\-Xmx2048m</java-opts>  
            <arg>$\{driveID\}</arg>  
        </java>  
        <ok to="merging"/>  
        <error to="fail"/>  
    </action>  
    <fork  name="merging">  
        <path start="mergeLidar"/>  
        <path start="mergeSignage"/>  
    </fork>  
    <action name='mergeLidar'>  
        <java>  
            <job-tracker>$\{jobTracker\}</job-tracker>  
            <name-node>$\{nameNode\}</name-node>  
            <configuration>  
                <property>  
                    <name>mapred.job.queue.name</name>  
                    <value>default</value>  
                </property>  
            </configuration>  
            <main-class>com.navteq.assetmgmt.hdfs.merge.MergerLoader</main-class>  
            <java-opts>\-Xmx2048m</java-opts>  
            <arg>\-drive</arg>  
            <arg>$\{driveID\}</arg>  
            <arg>\-type</arg>  
            <arg>Lidar</arg>  
            <arg>\-chunk</arg>  
            <arg>$\{lidarChunk\}</arg>  
        </java>  
    <ok  to="completed"/>  
    <error  to="fail"/>  
    </action>  
    <action name='mergeSignage'>  
        <java>  
            <job-tracker>$\{jobTracker\}</job-tracker>  
            <name-node>$\{nameNode\}</name-node>  
            <configuration>  
                <property>  
                    <name>mapred.job.queue.name</name>  
                    <value>default</value>  
                </property>  
            </configuration>  
            <main-class>com.navteq.assetmgmt.hdfs.merge.MergerLoader</main-class>  
            <java-opts>\-Xmx2048m</java-opts>  
            <arg>\-drive</arg>  
            <arg>$\{driveID\}</arg>  
            <arg>\-type</arg>  
            <arg>MultiCam</arg>  
            <arg>\-chunk</arg>  
            <arg>$\{signageChunk\}</arg>  
        </java>  
        <ok to="completed"/>  
        <error to="fail"/>  
    </action>  
    <join  name="completed" to="end"/>  
    <kill  name="fail">  
        <message>Java failed, error message\[$\{wf:errorMessage(wf:lastErrorNode())\}\]</message>  
    </kill>  
    <end  name='end'/>  
</workflow-app>

**代码1： 简单的Oozie工作流**

这个工作流定义了三个动作：ingestor、mergeLidar和mergeSignage。并把每个动作都实现为Map/Reduce\[4\]作业。这个工作流从start节点开始，然后把控制权交给Ingestor动作。一旦ingestor步骤完成，就会触发fork控制节点 \[4\]，它会并行地开始执行mergeLidar和mergeSignage\[5\]。这两个动作完成之后，就会触发join控制节点\[6\]。join节点成功完成之后，控制权就会传递给end节点，它会结束这个过程。

创建工作流之后，我们需要正确地对其进行部署。典型的Oozie部署是一个HDFS目录，其中包含workflow.xml(代码1)、config-default.xml和lib子目录，其中包含有工作流操作所要使用的类的jar文件。

![23d1866a0456294b.jpg][]

**        图1： Oozie部署**

config-default.xml文件是可选的，通常其中会包含对于所有工作流实例通用的工作流参数。代码2中显示的是config-default.xml的简单示例。

<configuration>  
    <property>  
        <name>jobTracker</name>  
        <value>sachicn003:2010</value>  
    </property>  
    <property>  
        <name>nameNode</name>  
        <value>hdfs://sachicn001:8020</value>  
    </property>  
    <property>  
        <name>queueName</name>  
        <value>default</value>  
    </property>  
</configuration>

**代码2： Config-default.xml**

完成了工作流的部署之后，我们可以使用Oozie提供的命令行工具\[5\]，它可以用于提交、启动和操作工作流。这个工具一般会运行在Hadoop簇集\[7\]的edge节点上，并需要一个作业属性文件(参见配置工作流属性)，见代码3。

oozie.wf.application.path=hdfs://sachicn001:8020/user/blublins/workflows/IPSIngestion  
jobTracker=sachicn003:2010  
nameNode=hdfs://sachicn001:8020

**代码3： 作业属性文件**

有了作业属性，我们就可以使用代码4中的命令来运行Oozie工作流。

oozie job –oozie http://sachidn002.hq.navteq.com:11000/oozie/ -D driveID=729-pp00002-2011-02-08-09-59-34 -D lidarChunk=4 -D signageChunk=20 -config job.properties –run

**　　列表4： 运行工作流命令**

**配置工作流属性**

在config-default.xml、作业属性文件和作业参数中有一些重叠，它们可以作为命令行调用的一部分传递给Oozie。尽管文档中没有清晰地指出何时使用哪个，但总体上的建议如下：

1.使用config-default.xml定义对于指定工作流从未改变过的参数。

2.对于给定的工作流部署通用的参数，建议使用作业属性。

3.对于指定的工作流调用特定的参数使用命令行参数。

Oozie处理这三种参数的方式如下：

1.使用所有命令行调用的参数

如果那里有任何无法解析的参数，那么就是用作业配置来解析

一旦所有其它方式都无法处理，那么就试着使用config-default.xm。

我们可以使用Oozie控制台(图2)来观察工作流执行的进程和结果。

![Oozie：配置工作流属性][Oozie]

**       图2： Oozie控制台**

我们还可以使用Oozie控制台来获得操作执行的细节，比方说作业的日志\[8\](图3)。

![Oozie：配置工作流属性][Oozie 1]

**        图3： Oozie控制台——作业日志**

**编程方式的工作流调用**

尽管上面所述的命令行界面能够很好地用于手动调用Oozie，但有时使用编程的方式调用Oozie更具有优势。当Oozie工作流是特定的应用程序或者大型企业过程的一部分，这就会很有用。我们可以使用Oozie Web Services APIs \[6\]或者Oozie Java client APIs \[7\]来实现这种编程方式的调用。代码5中展现的就是很简单的Oozie Java客户端的例子，它会触发上面描述的过程。

package com.navteq.assetmgmt.oozie;  
  
import java.util.LinkedList;  
import java.util.List;  
import java.util.Properties;  
  
import org.apache.oozie.client.OozieClient;  
import org.apache.oozie.client.OozieClientException;  
import org.apache.oozie.client.WorkflowJob;  
import org.apache.oozie.client.WorkflowJob.Status;  
  
public class WorkflowClient \{  
  
    privatestatic String OOZIE\_URL ="http://sachidn002.hq.navteq.com:11000/oozie/";  
    privatestatic String JOB\_PATH ="hdfs://sachicn001:8020/user/blublins/workflows/IPSIngestion";  
    privatestatic String JOB\_Tracker ="sachicn003:2010";  
    privatestatic String NAMENode ="hdfs://sachicn001:8020";  
  
    OozieClient wc =null;  
  
    public WorkflowClient(String url)\{  
        wc =new OozieClient(url);  
    \}  
  
    public String startJob(String wfDefinition, List<WorkflowParameter> wfParameters)  
        throws OozieClientException\{  
  
        // create a workflow job configuration and set the workflow application path  
        Properties conf = wc.createConfiguration();  
        conf.setProperty(OozieClient.APP\_PATH, wfDefinition);  
  
        // setting workflow parameters  
        conf.setProperty("jobTracker", JOB\_Tracker);  
        conf.setProperty("nameNode", NAMENode);  
        if((wfParameters !=null) && (wfParameters.size() >0))\{  
            for(WorkflowParameter parameter : wfParameters)  
                conf.setProperty(parameter.getName(), parameter.getValue());  
        \}  
        // submit and start the workflow job  
        return wc.run(conf);  
    \}  
  
    public Status getJobStatus(String jobID) throws OozieClientException\{  
        WorkflowJob job = wc.getJobInfo(jobID);  
        return job.getStatus();  
    \}  
  
    public staticvoid main(String\[\] args) throws OozieClientException, InterruptedException\{  
  
        // Create client  
        WorkflowClient client =new WorkflowClient(OOZIE\_URL);  
        // Create parameters  
        List<WorkflowParameter> wfParameters =new LinkedList<WorkflowParameter>();  
        WorkflowParameter drive =new WorkflowParameter("driveID","729-pp00004-2010-09-01-09-46");  
        WorkflowParameter lidar =new WorkflowParameter("lidarChunk","4");  
        WorkflowParameter signage =new WorkflowParameter("signageChunk","4");  
        wfParameters.add(drive);  
        wfParameters.add(lidar);  
        wfParameters.add(signage);  
        // Start Oozing  
        String jobId = client.startJob(JOB\_PATH, wfParameters);  
        Status status = client.getJobStatus(jobId);  
        if(status == Status.RUNNING)  
             System.out.println("Workflow job running");  
        else  
             System.out.println("Problem starting Workflow job");  
    \}  
\}

**代码5： 简单的Oozie Java客户端**

在此，我们首先使用Oozie服务器URL对工作流客户端进行初始化。初始化过程完成之后，我们就可以使用客户端提交并启动作业(startJob方法)，获得正在运行的作业的状态(getStatus方法)，以及进行其他操作。

**　　构建java动作，向工作流传递参数**

在之前的示例中，我们已经展示了如何使用标签向Java节点传递参数。由于Java节点是向Oozie引入自定义计算的主要方法，因此能够从Java节点向Oozie传递数据也同样重要。

根据Java节点的文档\[3\]，我们可以使用“capture-output””元素把Java节点生成的值传递回给Oozie上下文。然后，工作流的其它步骤可以通过EL-functions访问这些值。返回值需要以Java属性格式文件写出来。我们可以通过“JavaMainMapper.OOZIE\_JAVA\_MAIN\_CAPTURE\_OUTPUT\_FILE”常量从System属性中获得这些属性文件的名称。代码6是一个简单示例，演示了如何完成这项操作。

package com.navteq.oozie;  
import java.io.File;  
import java.io.FileOutputStream;  
import java.io.OutputStream;  
import java.util.Calendar;  
import java.util.GregorianCalendar;  
import java.util.Properties;  
  
public class GenerateLookupDirs \{  
  
    /\*\*  
    \* @param args  
    \*/  
    public staticfinallong dayMillis =1000\*60\*60\*24;  
    privatestatic final String OOZIE\_ACTION\_OUTPUT\_PROPERTIES ="oozie.action.output.properties";  
  
    public staticvoid main(String\[\] args) throws Exception \{  
        Calendar curDate =new GregorianCalendar();  
        int year, month, date;  
        String propKey, propVal;  
  
        String oozieProp = System.getProperty(OOZIE\_ACTION\_OUTPUT\_PROPERTIES);  
        if (oozieProp !=null) \{  
            File propFile =new File(oozieProp);  
            Properties props =new Properties();  
  
            for (int i =0; I <8; \++i) \{  
                year = curDate.get(Calendar.YEAR);  
                month = curDate.get(Calendar.MONTH) \+1;  
                date = curDate.get(Calendar.DATE);  
                propKey ="dir"\+i;  
                propVal = year \+"\-"\+  
                    (month <10?"0"\+ month : month) \+"\-"\+  
                    (date <10?"0"\+ date : date);  
                props.setProperty(propKey, propVal);  
                curDate.setTimeInMillis(curDate.getTimeInMillis() \- dayMillis);  
            \}  
            OutputStream os =new FileOutputStream(propFile);  
            props.store(os, "");  
            os.close();  
        \} else  
            thrownew RuntimeException(OOZIE\_ACTION\_OUTPUT\_PROPERTIES  
             \+" System property not defined");  
        \}  
\}

**代码6： 向Oozie传递参数**

在这个示例中，我们假设在HDFS中有针对每个日期的目录。这样，这个类首先会获得当前日期，然后再获得离现在最近的7个日期(包括今天)，然后把目录名称传递回给Oozie。

注：\[1\]edge节点是安装有Hadoop库的计算机，但不是真正簇集中的一部分。它是为能够连接到簇集中的应用程序所用的，并且会部署辅助服务以及能够直接访问簇集的最终用户应用程序。

\[2\]请参看Oozie安装的链接。

\[3\]这些作业的细节和本文无关，所以在其中没有描述。

\[4\]Map/Reduce作业能够以两种不同的方式在Oozie中实现——第一种是作为真正的Map/Reduce动作\[2\]，其中你会指定Mapper和Reducer类以及它们的配置信息；第二种是作为Java动作\[3\]，其中你会使用Hadoop API来指定启动Map/Reduce作业的类。因为我们所有的Java主函数都是使用Hadoop API，并且还实现了一些额外的功能，所以我们选择了第二种方法。

\[5\] Oozie确保两个动作会并行地提交给作业跟踪程序。在执行过程中实际的并行机制并不在Oozie的控制之内，并且依赖于作业的需求、簇集的能力以及Map/Reduce部署所使用的调度程序。

\[6\]join动作的功能是要同步fork动作启动的多个并行执行的线程。如果fork启动的所有执行的线程都能够成功完成，那么join动作就会等待它们全部完成。如果有至少一个线程执行失败，kill节点会“杀掉”剩余运行的线程。

\[7\] 这个节点不需要是安装了Oozie的计算机。

\[8\] Oozie的作业日志会包含工作流执行的细节，想要查看动作执行的细节，我们需要切换到Hadoop的Map/Reduce管理页面。

[23d1866a0456294b.jpg]: /images/20220311/160227bd655a4b66b5f04b43602cb8d2.png
[Oozie]: /images/20220311/4982f7cbeded4a268a227d6ac9aa0b70.png
[Oozie 1]: /images/20220311/7dac8f7d83014a448551a55bb9c6dbbe.png