oozie工作流相关入门整理

2023-01-01 19:47:39

Oozie支持工作流，其定义通过将多个Hadoop Job的定义按照一定的顺序组织起来，然后作为一个整体按照既定的路径运行。一个工作流已经定义了，通过启动该工作流Job，就会执行该工作流中包含的多个Hadoop Job，直到完成，这就是工作流Job的生命周期。

通过最简单的一个例子来了解oozie是如何运行的，运行oozie的服务器必须能够访问HDFS，可以提交hadoop mapreduce任务（如果需要执行hive，spark等任务，同样需要对应的环境）。

为了让实例比较简单，任务只是在提交后执行一次，并没有使用coordinator，只是使用了workflow.xml，在workflow中定义任务的整体流程，workflow.xml的定义内容简要如下：

<workflow-app xmlns="uri:oozie:workflow:0.4" name="${sys_name}-MonitorPlus-Task-${task_id}">

    <start to=“check-xxx-succ-flag"/>

    <decision name=“check-xxx-succ-flag">

        <switch>

<case to="check-mr-succ-flag">${fs:exists(concat(concat("/xxx/output/xxxList/",

     task_id),"/_SUCCESS"))}</case>

<default to=“do-xxx"/>

        </switch>

    </decision>

    <decision name="check-mr-succ-flag">

        <switch>

            <case to="end">${fs:exists(concat(concat(“/xxx/output/", task_id),"/_SUCCESS"))}</case>

            <default to="do-mr"/>

        </switch>

    </decision>

    <action name=“do-xxx" retry-max="3" retry-interval="10">

        <java>

            <job-tracker>${jobTracker}</job-tracker>

            <name-node>${namenode}</name-node>

            <configuration>

                <property>

                    <name>mapreduce.job.queuename</name>

                    <value>${queueName}</value>

                </property>

            </configuration>

           <main-class>com.xxx.Main</main-class>

         ...

类似于工作流，其中有开始节点 start，一些decision用于流程判断分支节点，以及具体的action，action中可以选择java, shell等类型节点，启动MapReduce任务或者可以选择跳转到某台服务器去启动shell脚本（理论上，如果可以启动shell脚本，就可以做任何在单机上可以做的事情）；分支节点可以根据一些条件，比如某个输出目录的_SUCCESS文件是否存在，将以前执行的某些阶段跳过，在执行action时可以设置重试次数以及重试间隔，以避免某次集群环境问题引起的失败。

工作流中可以根据需要设置特定的变量参数，以${varibleName}来定义，这样，使用统一的oozie定义来根据传递参数的不同来执行不同的任务。

工作流xml文件需要被放在HDFS上才能被oozie调度，如果在启动需要调度MapReduce任务，同样jar包也需要上传到hdfs上，形成下面的目录结构：

/xxx/oozie/workflow.xml

/xxx/oozie/lib

/xxx/oozie/lib/mr-1.7-SNAPSHOT-package.jar

/xxx/oozie/lib/xxx.mr-1.7-SNAPSHOT-package.jar

在workflow工作流定义的同级目录下，需要有一个lib目录，在lib目录中存在java节点MapReduce使用的jar包。需要注意的是，oozie并不是使用指定jar包的名称来启动任务的，在lib包中绝对不能存在某个jar包的不同版本，不能够出现多个相同主类。

工作流中设置的变量，需要通过一个叫做job.properties的配置文件来填充（不能改名称，如果需要多个，请放到不同的文件夹下），job.properties需要配置下面的参数：

namenode=hdfs://ns1

jobTracker=xxx.cn:8032

oozie_url=http://xxx.cn:11000/oozie

oozie.wf.application.path=/xxx/oozie/workflow.xml

必须要求设置这三个参数namenode，jobTracker，oozie_url，以及任务workflow.xml在hdfs上的位置。其他的参数，视具体的workflow.xml而定，启动oozie的方式为：

oozie job -oozie ${oozie_url} -config ${job_cfg} -run

如果oozie是在本机运行，不需要-oozie ${oozie_url}，任务提交成功后，就会返回一个oozie任务id，可以在oozie web url中查看该任务的具体执行情况。

以上是通过命令行来调用oozie，当然也可以通过java oozie client来调用oozie执行，需要依赖pom：

<dependency>

    <groupId>org.apache.oozie</groupId>

    <artifactId>oozie-client</artifactId>

    <version>4.1.0</version>

</dependency>

可以参考oozie官网中介绍的例子来完成任务提交，这里参照之前的例子，用java实现了一遍：

public class WorkflowClient {

    public static final String OOZIE_URL = "http://xxx.cn:11000/oozie";

    private OozieClient oozieClient = new OozieClient(OOZIE_URL);

    public WorkflowClient() {

    }

    /**

     * @param workflowDefinition

     * @param workflowParameters

     * @return - jobId

     * @throws org.apache.oozie.client.OozieClientException

     */

    public String startJob(Properties properties)

            throws OozieClientException {

        // create a workflow job configuration and set the workflow application path

        Properties configuration = oozieClient.createConfiguration();

        Enumeration<?> enumeration = properties.propertyNames();

        while (enumeration.hasMoreElements()) {

            Object element = enumeration.nextElement();

            configuration.setProperty(element.toString(), properties.getProperty(element.toString()).toString());

        }

        return oozieClient.run(configuration);

    }

    public static void main(String[] args) throws OozieClientException, InterruptedException, IOException {

        String jobPropertyFile = args[0];

        // Create client

        WorkflowClient client = new WorkflowClient();

        // Create parameters

        Properties properties = new Properties();

        properties.load(new FileInputStream(new File(jobPropertyFile)));

        // Start Oozing

        String jobId = client.startJob(properties);

        SysOutLogger.info("jobId: " + jobId);

        WorkflowJob jobInfo = client.oozieClient.getJobInfo(jobId);

        SysOutLogger.info("job url: " + jobInfo.getConsoleUrl());

        while (true) {

            Thread.sleep(1000L);

            WorkflowJob.Status status = client.oozieClient.getJobInfo(jobId).getStatus();

            if (status == WorkflowJob.Status.SUCCEEDED || status == WorkflowJob.Status.FAILED

|| status == WorkflowJob.Status.KILLED) {

                SysOutLogger.info("Job finish with status: " + status);

                break;

            } else {

                SysOutLogger.info("job running: " + jobInfo.getStatus());

            }

        }

    }

}

用maven打包完成后，将其放到服务器端运行:

[INFO]  2015-04-16 11:02:18 : jobId: 0000002-150415180413953-oozie-supe-W

[INFO]  2015-04-16 11:02:18 : job url: http://xxx.cn:11000/oozie?job=0000002-150415180413953-oozie-supe-W

[INFO]  2015-04-16 11:02:18 : job running: RUNNING

[INFO]  2015-04-16 11:02:19 : Job finish with status: SUCCEEDED

在job提交后，根据jobId，就通过oozieClient获得其WorkflowJob，使用OozieClient对应的API还可以对已经执行的任务进行kill/rerun等操作。由于没有找到ooze对应的事件API，这里是通过定时获取状态的方式将状态打印出来，直到任务完成。从oozie文档中可以查找到，oozie是可以和JMS服务集成，将任务执行状态发送到对应topic上，这样就可以依赖JMS Listener实现类似的事件监听/通知机制。

码农公寓

相关文章