Oozie工作流分析

时间:2022-07-22
本文章向大家介绍Oozie工作流分析，主要内容包括其使用实例、应用技巧、基本知识点总结和需要注意事项，具有一定的参考价值，需要的朋友可以参考一下。

我们在实际的生成操作中经常需要将一些任务在晚上开启进行定时执行，或者多个作业，例如hive,mapreduce,shell等任务的组合调用。

我们可以使用linux的contab + spervisor + inotify-tool进行任务的配值，但是操作起来麻烦，而且没有可视化的参数返回界面。也不能保证任务的可靠性。

在hadoop技术栈中我们可以使用oozie做为任务的调度与定时触发的工具。可以方便的帮助我们进行管理和调度我们常见的9中作业调度。

oozie基本架构

是一个工作流调度系统

  • 工作流的调度是DAG（有向无环图）-Directed Acyclical Graphs
  • Coordinator job可以通过时间和数据集的可用性触发
  • 集成了Hadoop生态系统的其它任务，如mr,pig,hive,sqoop,distcp
  • 可扩展：一个Oozie就是一个mr程序，但是仅仅是map，没有reduce 可靠性：任务失败后的重试

oozie.png

workflow，coordinator，bundle

Workflow：

   工作流，由我们需要处理的每个工作组成，进行需求的流式处理。

Coordinator：

   协调器，可以理解为工作流的协调器，可以将多个工作流协调成一个工作流来进行处理。

Bundle：

   捆，束。将一堆的coordinator进行汇总处理。

简单来说，workflow是对要进行的顺序化工作的抽象，coordinator是对要进行的顺序化的workflow的抽象，bundle是对一堆coordiantor的抽象。层级关系层层包裹。

Oozie的作业有三部分组成，分别是job.properties，workflow.xml，lib文件夹。

Job.properties

配值需要的参数

nameNode hdfs地址

jobTracker jobTracker（ResourceManager）地址

queueName Oozie队列（默认填写default）

examplesRoot 全局目录（默认填写examples）

oozie.usr.system.libpath 是否加载用户lib目录（true/false）

oozie.libpath 用户lib库所在的位置

oozie.wf.application.path

Oozie流程所在hdfs地址（workflow.xml所在的地址）

user.name 当前用户

Coordinator：oozie.coord.application.path

Coordinator.xml地址（没有可以不写）

Bundle：oozie.bundle.application.path

Bundle.xml地址（没有可以不写）

nameNode=hdfs://cm1:8020

jobTracker=cm1:8032

queueName=default

examplesRoot=examples

oozie.wf.application.path=${nameNode}/user/workflow/oozie/shell
workflow.xml
<workflow-app xmlns="uri:oozie:workflow:0.4" name="${sys_name}-MonitorPlus-Task-${task_id}">  

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

3.    

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

5.          <switch>  

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

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

8.  <default to=“do-xxx"/>  

9.          </switch>  

10.     </decision>  

11.   

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

13.         <switch>  

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

15.             <default to="do-mr"/>  

16.         </switch>  

17.     </decision>  

18.   

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

20.         <java>  

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

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

23.             <configuration>  

24.                 <property>  

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

26.                     <value>${queueName}</value>  

27.                 </property>  

28.             </configuration>  

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

30.          </java>

31.        <ok to=”end”/>

32.        <error to=”fail”/>

33.     </action>

34.     <kill name=”fail”>

35.         <message>Map/Reduce failed.error message[${wf:errorMessage(wf:lastErrorNode())}]</message>

36.     </kill>

37.     <end name=”end”/>

38. </workflow-app>

Oozie的节点分成两种，流程控制节点和动作节点。所谓的节点实际就是一组标签。两种节点分别如下：

流程控制节点

<start />——定义workflow的开始

<end />——定义workflow的结束

<decision />——实现switch功能

<switch><case /><default /></switch>标签连用

<sub-workflow>——调用子workflow

<kill />——程序出错后跳转到这个节点执行相关操作

<fork />——并发执行workflow

<join />——并发执行结束（与fork一起使用）

动作节点

<shell />——表示运行的是shell操作

<java />——表示运行的java程序

<fs />——表示是对hdfs进行操作

<MR />——表示进行的是MR操作

<hive />——表示进程的是hive操作

<sqoop />——表示进行的是sqoop的相关操作

文件需要被放在HDFS上才能被oozie调度，如果在启动需要调动MR任务，jar包同样需要在hdfs上。

lib文件夹

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

oozie cli

l 启动任务

oozie job -oozie oozie_url -config job.properties_address-run

l 停止任务

oozie job -oozie oozie_url -kill jobId -oozie-oozi-W

l 提交任务

oozie job -oozie oozie_url -config job.properties_address -submit

l 开始任务

oozie job -oozie oozie_url -config job.properties_address -startJobId -oozie-oozi-W

l 查看任务执行情况

oozie job -oozieoozie_url -config job.properties_address -info jobId -oozie-oozi-W

说明： 所有的命令都是以oozie job -oozie oozie_url 开头的-config 制定job.properties文件夹的位置，-run 文件启动后会返回一个唯一的jobId，供之后使用。

Oozie Coordinator job 定时任务

修改时区 .修改 core-site.xml

<property>
    <name>oozie.processing.timezone</name>
    <value>GMT+0800</value>
</property>

修改 $OOZIE_HOME/oozie-server/webapps/oozie/oozie-console.js

function getTimeZone() {
Ext.state.Manager.setProvider(new Ext.state.CookieProvider());
return Ext.state.Manager.get("TimezoneId","GMT+0800");
}

1.编辑job.properties

nameNode=hdfs://cen-ubuntu.cenzhongman.com:8020
jobTracker=localhost:8032
queueName=default
oozieAppsRoot=oozie-apps

oozie.coord.application.path=${nameNode}/user/cen/${oozieAppsRoot}/cron-schedule
start=2017-07-30T14:40+0800
end=2017-07-30T14:59+0800
workflowAppUri=${nameNode}/user/cen/${oozieAppsRoot}/cron-schedule

2.编辑 workflow.xml 文件

<workflow-app xmlns="uri:oozie:workflow:0.5" name="no-op-wf">
    <start to="end"/>
    <end name="end"/>
</workflow-app>

3.编辑coordinator.xml 文件

<coordinator-app name="cron-coord" frequency="0/1 * * * *" start="${start}" end="${end}" timezone="GMT+0800"
                 xmlns="uri:oozie:coordinator:0.4">
        <action>
        <workflow>
            <app-path>${workflowAppUri}</app-path>
            <configuration>
                <property>
                    <name>jobTracker</name>
                    <value>${jobTracker}</value>
                </property>
                <property>
                    <name>nameNode</name>
                    <value>${nameNode}</value>
                </property>
                <property>
                    <name>queueName</name>
                    <value>${queueName}</value>
                </property>
            </configuration>
        </workflow>
    </action>
</coordinator-app>

4.上传文件至 HDFS 5.执行任务

export OOZIE_URL=http://cen-ubuntu:11000/oozie/
bin/oozie job --config oozie-apps/cron-schedule/job.properties -run

oozie JAVA API

直接将oozie下的oozie-client.jar包拷贝带eclipse中,就可以使用java进行启动oozie任务了，这也方便了项目的集成。

public class UserProxy {
 
    public static void main(String[] args) throws Exception {
        HadoopLogin login = new HadoopLogin();
        final Configuration conf = login.loginHdfs("ibdc","C:/Program Files (x86)/Java/newhadoop_oozieweb_conf/ibdc.keytab");
        
        UserGroupInformation.getLoginUser().doAs(new PrivilegedExceptionAction(){
 
            public Void run() throws Exception {
                submitJob();
                return null;
            }
            
        });

    }
    
    private static void submitJob() throws OozieClientException, InterruptedException
    {
         // get a OozieClient for local Oozie
        XOozieClient  wc =new AuthOozieClient("http://hadoop7:11000/oozie/");
//       OozieClient wc = new OozieClient("http://hadoop7:11000/oozie/v1/job/");
//       AuthOozieClient wc = new AuthOozieClient("http://hadoop7:11000/oozie/", AuthOozieClient.AuthType.KERBEROS.toString());
        try {
            System.out.println(UserGroupInformation.getLoginUser());
        } catch (IOException e) {
            e.printStackTrace();
        }
         // create a workflow job configuration and set the workflow application path
         Properties conf = wc.createConfiguration();
         conf.setProperty(OozieClient.APP_PATH, "hdfs://nameservice1/user/oozieweb/oozie-app/oozieweb/workflow/antest2");
        
         // setting workflow parameters
         conf.setProperty("jobTracker", "hadoop7:8032");
         conf.setProperty("nameNode", "hdfs://nameservice1");
//       conf.setProperty("examplesRoot", EXAMPLE_DIR);
         conf.setProperty("queueName", "cdrapp");
//       conf.setProperty("outputDir", OUTPUT_DIR);
//       conf.setProperty("oozie.wf.rerun.failnodes", "true");
         conf.setProperty("hdfs.keytab.file", "C:/Program Files (x86)/Java/newhadoop_oozieweb_conf/oozieweb.keytab");
         conf.setProperty("hdfs.kerberos.principal", "oozieweb"); 
         conf.setProperty("mapred.mapper.new-api", "true");
         conf.setProperty("mapred.reducer.new-api", "true");
         conf.setProperty("oozie.use.system.libpath", "true");
    
         // submit and start the workflow job
         String jobId = wc.run(conf);
         System.out.println("Workflow job submitted");
         
         // wait until the workflow job finishes printing the status every 10 seconds
         while (wc.getJobInfo(jobId).getStatus() == WorkflowJob.Status.RUNNING) {
            System.out.println("Workflow job running ...");
            Thread.sleep(10 * 1000);
         }
         
         // print the final status of the workflow job
         System.out.println("Workflow job completed ...");
         System.out.println(wc.getJobInfo(jobId));
    }

}

oozie和azkaban 区别

工作流定义：Oozie是通过xml定义的而Azkaban为properties来定义。

部署过程： Oozie的部署太虐心了。有点难。同时它是从Yarn上拉任务日志。

Azkaban中如果有任务出现失败，只要进程有效执行，那么任务就算执行成功，这是BUG，但是Oozie能有效的检测任务的成功与失败。

操作工作流：Azkaban使用Web操作。Oozie支持Web，RestApi，Java API操作。

权限控制： Oozie基本无权限控制，Azkaban有较完善的权限控制，入用户对工作流读写执行操作。

Oozie的action主要运行在hadoop中而Azkaban的actions运行在Azkaban的服务器中。

记录workflow的状态：Azkaban将正在执行的workflow状态保存在内存中，Oozie将其保存在Mysql中。

出现失败的情况：Azkaban会丢失所有的工作流，但是Oozie可以在继续失败的工作流运行。

随机文章
本站知识点必读
最近更新