Oozie开发指南

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

Oozie提出了Coordinator的概念,能够将每个工作流的Job作为一个Action来运行。相当于工作流中的一个执行节点。这样就能够将多个工作流Job组织起来,称为Coordinator Job,并制定触发时间和频率,还可以配置数据集、并发数等。一个Coordinator Job包含了在Job外部设置执行周期和频率的语义。类似于在工作流外部专家了一个协调器来管理这些工作流的工作流Job的运行。

如果在集群安装了Hue,也可以通过页面操作配置工作流,具体操作步骤点此查看。以下介绍通过后台配置工作流的方法:

1. 运行Coordinator Job

先看一下官方发行包自带的一个简单例子 oozie/examples/src/main/apps/cron。它能够实现定时调度一个工作流Job运行,这个例子中给出的一个空的工作流Job,也是为了演示能够使用Coordinator系统给调度起来。

点击下载oozie-example.tar.gz

这个例子有3个配置文件。修改后分别如下所示:

  • job.properties配置
nameNode=hdfs://uhadoop-XXXXXX-master1:8020
jobTracker=uhadoop-XXXXXX-master1:23140
queueName=default
examplesRoot=examples
 
oozie.coord.application.path=${nameNode}/user/${user.name}/${examplesRoot}/apps/cron
start=2016-12-01T19:00Z
end=2016-12-31T01:00Z
workflowAppUri=${nameNode}/user/${user.name}/${examplesRoot}/apps/cron

修改了Hadoop集群的配置,以及调度起止时间范围。

  • wordflow.xml
 xmlns="uri:oozie:workflow:0.5" name="one-op-wf">
     to="action1"/>
     name="action1">
        />
     to="end"/>
     to="end"/>
    >
     name="end"/>
>

这是一个空Job,没做任何修改。

  • corrdinator.xml配置
 name="cron-coord" frequency="${coord:minutes(2)}" start="${start}" end="${end}" timezone="UTC" xmlns="uri:oozie:coordinator:0.2">
     >
          >
               >${workflowAppUri}>
               >
                    >
                         >jobTracker>
                         >${jobTracker}>
                    >
                    >
                         >nameNode>
                         >${nameNode}>
                    >
                    >
                         >queueName>
                         >${queueName}>
                    >
               >
          >
     >
>

将定时调度频率改为2分钟。然后,将这3个文件上传到HDFS上。

启动一个Coordinator Job和启动一个Oozie工作流Job类似,执行如下命令即可:

bin/oozie job -oozie http://uhadoop-XXXXXX-master2:11000/oozie -config /home/hadoop/oozie/examples/src/main/apps/cron/job.properties -run

运行上面命令,在控制台上会返回这个Job的ID,我们也可以通过Oozie的Web控制台来查看。

2. Coordinator Action

一个Coordinator Job会创建并执行Coordinator Action。通常一个Coordinator Action是一个工作流Job,这个工作流Job会生成一个dataset实例并处理这个数据集。当一个Coordinator Action被创建以后,它会一直等待满足执行条件的所有输入事件的完成然后执行,或者发生超时。

每个Coordinator Job都有一个驱动事件,来决定它所包含的Coordinator Action的初始化。对于同步Coordinator Job来说,触发执行频率就是一个驱动事件。同样,组成Coordinator Job的基本单元是Coordinator Action,它不像Oozie工作流Job只有OK和Error两个执行结果,一个Coordinator 动作的状态集合,如下所示:

WAITING
READY
SUBMITTED
TIMEDOUT
RUNNING
KILLED
SUCCEEDED
FAILED

3. Coordinator Application

Coordinator Application当满足一定条件时,会触发Oozie工作流。其中,触发条件可以是一个时间频率、一个dataset实例是否可用,或者可能是外部的其他事件。 Coordinator Job是一个Coordinator应用的运行实例,这个Coordinator Job是在Oozie提供的Coordinator引擎上运行的,并且这个实例从指定的时间开始,直到运行结束。一个Coordinator Job具有以下几个状态:

PREP
RUNNING
RUNNINGWITHERROR
PREPSUSPENDED
SUSPENDED
SUSPENDEDWITHERROR
PREPPAUSED
PAUSED
PAUSEDWITHERROR
SUCCEEDED
DONEWITHERROR
KILLED
FAILED

Coordinator Job的状态比一个基本的Oozie工作流Job的状态要复杂的多。因为Coordinator Job的基本执行单元可能是一个基本Oozie Job,而且外加了一些调度信息,必然要增加额外的状态来描述。

4. Coordinator Application定义

一个同步的Coordinator Appliction定义的语法格式,如下所示:

 name="[NAME]" frequency="[FREQUENCY]" start="[DATETIME]" end="[DATETIME]" timezone="[TIMEZONE]" xmlns="uri:oozie:coordinator:0.1">
     >
          >[TIME_PERIOD]>
          >[CONCURRENCY]>
          >[EXECUTION_STRATEGY]>
     >
     >
          >[SHARED_DATASETS]>
          ...
          
           name="[NAME]" frequency="[FREQUENCY]" initial-instance="[DATETIME]" timezone="[TIMEZONE]">
               >[URI_TEMPLATE]>
          >
          ...
     >
     >
           name="[NAME]" dataset="[DATASET]">
               >[INSTANCE]>
               ...
          >
          ...
           name="[NAME]" dataset="[DATASET]">
               >[INSTANCE]>
               >[INSTANCE]>
          >
          ...
     >
     >
           name="[NAME]" dataset="[DATASET]">
               >[INSTANCE]>
          >
          ...
     >
     >
          >
               >[WF-APPLICATION-PATH]>
               >
                    >
                         >[PROPERTY-NAME]>
                         >[PROPERTY-VALUE]>
                    >
                    ...
               >
          >
     >
>

基于上述定义语法格式,我们分别说明对应元素的含义,如下所示:

  • control元素
元素名称 含义说明
timeout 超时时间,单位为分钟。当一个Coordinator Job启动的时候,会初始化多个Coordinator动作,timeout用来限制这个初始化过程。默认值为-1,表示永远不超时,如果为0 则总是超时。
concurrency 并发数,指多个Coordinator Job并发执行,默认值为1。
execution 配置多个Coordinator Job并发执行的策略:默认是FIFO。另外还有两种:LIFO(最新的先执行)、LAST_ONLY(只执行最新的Coordinator Job,其它的全部丢弃)。
throttle 一个Coordinator Job初始化时,允许Coordinator动作处于WAITING状态的最大数量。
  • dataset元素

Coordinator Job中有一个Dataset的概念,它可以为实际计算提供计算的数据,主要是指HDFS上的数据目录或文件,能够配置数据集生成的频率(Frequency)、URI模板、时间等信息,下面看一下dataset的语法格式:

 name="[NAME]" frequency="[FREQUENCY]" initial-instance="[DATETIME]" timezone="[TIMEZONE]">
     >[URI TEMPLATE]>
     >[FILE NAME]>
>
举例如下:
 name="stats_hive_table" frequency="${coord:days(1)}" initial-instance="2016-12-25T00:00Z" timezone="America/Los_Angeles">
     >
          hdfs://m1:9000/hive/warehouse/user_events/${YEAR}${MONTH}/${DAY}/data
     >
     >donefile.flag>
>

上面会每天都会生成一个用户事件表,可以供Hive查询分析,这里指定了这个数据集的位置,后续计算会使用这部分数据。其中,uri-template指定了一个匹配的模板,满足这个模板的路径都会被作为计算的基础数据。 另外,还有一种定义dataset集合的方式,将多个dataset合并成一个组来定义,语法格式如下所示:

>
     >[SHARED_DATASETS]>
     ...
      name="[NAME]" frequency="[FREQUENCY]" initial-instance="[DATETIME]" timezone="[TIMEZONE]">
          >[URI TEMPLATE]>
     >
     ...
>
  • input-events和output-events元素

一个Coordinator Application的输入事件指定了要执行一个Coordinator动作必须满足的输入条件,在Oozie当前版本,只支持使用dataset实例。

一个Coordinator Action可能会生成一个或多个dataset实例,在Oozie当前版本,输出事件只支持输出dataset实例。

常量表示形式 含义说明
${coord:minutes(int n)} 返回日期时间:从一开始,周期执行n分钟
${coord:hours(int n)} 返回日期时间:从一开始,周期执行n * 60分钟
${coord:days(int n)} 返回日期时间:从一开始,周期执行n * 24 * 60分钟
${coord:months(int n)} 返回日期时间:从一开始,周期执行n * M * 24 * 60分钟(M表示一个月的天数)
${coord:endOfDays(int n)} 返回日期时间:从当天的最晚时间(即下一天)开始,周期执行n * 24 * 60分钟
${coord:endOfMonths(1)} 返回日期时间:从当月的最晚时间开始(即下个月初),周期执行n * 24 * 60分钟
${coord:current(int n)} 返回日期时间:从一个Coordinator动作(Action)创建时开始计算,第n个dataset实例执行时间
${coord:dataIn(String name)} 在输入事件(input-events)中,解析dataset实例包含的所有的URI
${coord:dataOut(String name)} 在输出事件(output-events)中,解析dataset实例包含的所有的URI
${coord:offset(int n, String timeUnit)} 表示时间偏移,如果一个Coordinator动作创建时间为T,n为正数表示向时刻T之后偏移,n为负数向向时刻T之前偏移,timeUnit表示时间单位(选项有MINUTE、HOUR、DAY、MONTH、YEAR)
${coord:hoursInDay(int n)} 指定的第n天的小时数,n>0表示向后数第n天的小时数,n=0表示当天小时数,n<0表示向前数第n天的小时数
${coord:daysInMonth(int n)} 指定的第n个月的天数,n>0表示向后数第n个月的天数,n=0表示当月的天数,n<0表示向前数第n个月的天数
${coord:tzOffset()} dataset对应的时区与Coordinator Job的时区所差的分钟数
${coord:latest(int n)} 最近以来,当前可以用的第n个dataset实例
${coord:future(int n, int limit)} 当前时间之后的dataset实例,n>=0,当n=0时表示立即可用的dataset实例,limit表示dataset实例的个数
${coord:nominalTime()} nominal时间等于Coordinator Job启动时间,加上多个Coordinator Job的频率所得到的日期时间。例如:start=”2009-01-01T24:00Z”,end=”2009-12-31T24:00Z”,frequency=”${coord:days(1)}”,frequency=”${coord:days(1)},则nominal时间为:2009-01-02T00:00Z、2009-01-03T00:00Z、2009-01-04T00:00Z、…、2010-01-01T00:00Z
${coord:actualTime()} Coordinator动作的实际创建时间。例如:start=”2011-05-01T24:00Z”,end=”2011-12-31T24:00Z”,frequency=”${coord:days(1)}”,则实际时间为:2011-05-01,2011-05-02,2011-05-03,…,2011-12-31
${coord:user()} 启动当前Coordinator Job的用户名称
${coord:dateOffset(String baseDate, int instance, String timeUnit)} 计算新的日期时间的公式:newDate = baseDate + instance * timeUnit,如:baseDate=’2009-01-01T00:00Z’,instance=’2′,timeUnit=’MONTH’,则计算得到的新的日期时间为’2009-03-01T00:00Z’。
${coord:formatTime(String timeStamp, String format)} 格式化时间字符串,format指定模式。