摘要：開(kāi)發(fā)指南開(kāi)發(fā)指南開(kāi)發(fā)指南本篇目錄運(yùn)行運(yùn)行定義定義定義工作流，為將多個(gè)按照一定的順序組織起來(lái)，按照既定的路徑運(yùn)行的一個(gè)整體。配置將定時(shí)調(diào)度頻率改為分鐘。一個(gè)會(huì)創(chuàng)建并執(zhí)行。例如，，，，則時(shí)間為動(dòng)作的實(shí)際創(chuàng)建時(shí)間。

Oozie開(kāi)發(fā)指南

本篇目錄

1. 運(yùn)行Coordinator Job2. Coordinator Action3. Coordinator Application4. Coordinator Application定義

Oozie定義工作流，為將多個(gè)Hadoop Job按照一定的順序組織起來(lái)，按照既定的路徑運(yùn)行的一個(gè)整體。通過(guò)啟動(dòng)工作流，就會(huì)執(zhí)行該工作流中的多個(gè)Hadoop Job，直到完成。這就是工作流的生命周期。

Oozie提出了Coordinator的概念，能夠?qū)⒚總€(gè)工作流的Job作為一個(gè)Action來(lái)運(yùn)行。相當(dāng)于工作流中的一個(gè)執(zhí)行節(jié)點(diǎn)。這樣就能夠?qū)⒍鄠€(gè)工作流Job組織起來(lái)，稱為Coordinator Job，并制定觸發(fā)時(shí)間和頻率，還可以配置數(shù)據(jù)集、并發(fā)數(shù)等。一個(gè)Coordinator Job包含了在Job外部設(shè)置執(zhí)行周期和頻率的語(yǔ)義。類似于在工作流外部專家了一個(gè)協(xié)調(diào)器來(lái)管理這些工作流的工作流Job的運(yùn)行。

如果在集群安裝了Hue，也可以通過(guò)頁(yè)面操作配置工作流，具體操作步驟點(diǎn)此查看。以下介紹通過(guò)后臺(tái)配置工作流的方法：

1. 運(yùn)行Coordinator Job

先看一下官方發(fā)行包自帶的一個(gè)簡(jiǎn)單例子 oozie/examples/src/main/apps/cron。它能夠?qū)崿F(xiàn)定時(shí)調(diào)度一個(gè)工作流Job運(yùn)行，這個(gè)例子中給出的一個(gè)空的工作流Job，也是為了演示能夠使用Coordinator系統(tǒng)給調(diào)度起來(lái)。

點(diǎn)擊下載oozie-example.tar.gz

這個(gè)例子有3個(gè)配置文件。修改后分別如下所示：

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/cronCopyErrorSuccess

修改了Hadoop集群的配置，以及調(diào)度起止時(shí)間范圍。

wordflow.xml

<workflow-app xmlns="uri:oozie:workflow:0.5" name="one-op-wf">
    <start to="action1"/>
    <action name="action1">
        <fs/>
    <ok to="end"/>
    <error to="end"/>
    action>
    <end name="end"/>
workflow-app>CopyErrorSuccess

這是一個(gè)空J(rèn)ob，沒(méi)做任何修改。

corrdinator.xml配置

<coordinator-app name="cron-coord" frequency="${coord:minutes(2)}" start="${start}" end="${end}" timezone="UTC" xmlns="uri:oozie:coordinator:0.2">
     <action>
          <workflow>
               <app-path>${workflowAppUri}app-path>
               <configuration>
                    <property>
                         <name>jobTrackername>
                         <value>${jobTracker}value>
                    property>
                    <property>
                         <name>nameNodename>
                         <value>${nameNode}value>
                    property>
                    <property>
                         <name>queueNamename>
                         <value>${queueName}value>
                    property>
               configuration>
          workflow>
     action>
coordinator-app>CopyErrorSuccess

將定時(shí)調(diào)度頻率改為2分鐘。然后，將這3個(gè)文件上傳到HDFS上。

啟動(dòng)一個(gè)Coordinator Job和啟動(dòng)一個(gè)Oozie工作流Job類似，執(zhí)行如下命令即可：

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

運(yùn)行上面命令，在控制臺(tái)上會(huì)返回這個(gè)Job的ID，我們也可以通過(guò)Oozie的Web控制臺(tái)來(lái)查看。

2. Coordinator Action

一個(gè)Coordinator Job會(huì)創(chuàng)建并執(zhí)行Coordinator Action。通常一個(gè)Coordinator Action是一個(gè)工作流Job，這個(gè)工作流Job會(huì)生成一個(gè)dataset實(shí)例并處理這個(gè)數(shù)據(jù)集。當(dāng)一個(gè)Coordinator Action被創(chuàng)建以后，它會(huì)一直等待滿足執(zhí)行條件的所有輸入事件的完成然后執(zhí)行，或者發(fā)生超時(shí)。

每個(gè)Coordinator Job都有一個(gè)驅(qū)動(dòng)事件，來(lái)決定它所包含的Coordinator Action的初始化。對(duì)于同步Coordinator Job來(lái)說(shuō)，觸發(fā)執(zhí)行頻率就是一個(gè)驅(qū)動(dòng)事件。同樣，組成Coordinator Job的基本單元是Coordinator Action，它不像Oozie工作流Job只有OK和Error兩個(gè)執(zhí)行結(jié)果，一個(gè)Coordinator 動(dòng)作的狀態(tài)集合，如下所示：

WAITING

READY

SUBMITTED

TIMEDOUT

RUNNING

KILLED

SUCCEEDED

FAILED

3. Coordinator Application

Coordinator Application當(dāng)滿足一定條件時(shí)，會(huì)觸發(fā)Oozie工作流。其中，觸發(fā)條件可以是一個(gè)時(shí)間頻率、一個(gè)dataset實(shí)例是否可用，或者可能是外部的其他事件。 Coordinator Job是一個(gè)Coordinator應(yīng)用的運(yùn)行實(shí)例，這個(gè)Coordinator Job是在Oozie提供的Coordinator引擎上運(yùn)行的，并且這個(gè)實(shí)例從指定的時(shí)間開(kāi)始，直到運(yùn)行結(jié)束。一個(gè)Coordinator Job具有以下幾個(gè)狀態(tài)：

PREP

RUNNING

RUNNINGWITHERROR

PREPSUSPENDED

SUSPENDED

SUSPENDEDWITHERROR

PREPPAUSED

PAUSED

PAUSEDWITHERROR

SUCCEEDED

DONEWITHERROR

KILLED

FAILED

Coordinator Job的狀態(tài)比一個(gè)基本的Oozie工作流Job的狀態(tài)要復(fù)雜的多。因?yàn)镃oordinator Job的基本執(zhí)行單元可能是一個(gè)基本Oozie Job，而且外加了一些調(diào)度信息，必然要增加額外的狀態(tài)來(lái)描述。

4. Coordinator Application定義

一個(gè)同步的Coordinator Appliction定義的語(yǔ)法格式，如下所示：

<coordinator-app name="[NAME]" frequency="[FREQUENCY]" start="[DATETIME]" end="[DATETIME]" timezone="[TIMEZONE]" xmlns="uri:oozie:coordinator:0.1">
     <controls>
          <timeout>[TIME_PERIOD]timeout>
          <concurrency>[CONCURRENCY]concurrency>
          <execution>[EXECUTION_STRATEGY]execution>
     controls>
     <datasets>
          <include>[SHARED_DATASETS]include>
          ...
          
          <dataset name="[NAME]" frequency="[FREQUENCY]" initial-instance="[DATETIME]" timezone="[TIMEZONE]">
               <uri-template>[URI_TEMPLATE]uri-template>
          dataset>
          ...
     datasets>
     <input-events>
          <data-in name="[NAME]" dataset="[DATASET]">
               <instance>[INSTANCE]instance>
               ...
          data-in>
          ...
          <data-in name="[NAME]" dataset="[DATASET]">
               <start-instance>[INSTANCE]start-instance>
               <end-instance>[INSTANCE]end-instance>
          data-in>
          ...
     input-events>
     <output-events>
          <data-out name="[NAME]" dataset="[DATASET]">
               <instance>[INSTANCE]instance>
          data-out>
          ...
     output-events>
     <action>
          <workflow>
               <app-path>[WF-APPLICATION-PATH]app-path>
               <configuration>
                    <property>
                         <name>[PROPERTY-NAME]name>
                         <value>[PROPERTY-VALUE]value>
                    property>
                    ...
               configuration>
          workflow>
     action>
coordinator-app>CopyErrorSuccess

基于上述定義語(yǔ)法格式，我們分別說(shuō)明對(duì)應(yīng)元素的含義，如下所示：

control元素 元素名稱含義說(shuō)明timeout超時(shí)時(shí)間，單位為分鐘。當(dāng)一個(gè)CoordinatorJob啟動(dòng)的時(shí)候，會(huì)初始化多個(gè)Coordinator動(dòng)作，timeout用來(lái)限制這個(gè)初始化過(guò)程。默認(rèn)值為-1，表示永遠(yuǎn)不超時(shí)，如果為0則總是超時(shí)。concurrency并發(fā)數(shù)，指多個(gè)CoordinatorJob并發(fā)執(zhí)行，默認(rèn)值為1。execution配置多個(gè)CoordinatorJob并發(fā)執(zhí)行的策略：默認(rèn)是FIFO。另外還有兩種：LIFO（最新的先執(zhí)行）、LAST_ONLY（只執(zhí)行最新的CoordinatorJob，其它的全部丟棄）。throttle一個(gè)CoordinatorJob初始化時(shí)，允許Coordinator動(dòng)作處于WAITING狀態(tài)的最大數(shù)量。 dataset元素

Coordinator Job中有一個(gè)Dataset的概念，它可以為實(shí)際計(jì)算提供計(jì)算的數(shù)據(jù)，主要是指HDFS上的數(shù)據(jù)目錄或文件，能夠配置數(shù)據(jù)集生成的頻率（Frequency）、URI模板、時(shí)間等信息，下面看一下dataset的語(yǔ)法格式：

<dataset name="[NAME]" frequency="[FREQUENCY]" initial-instance="[DATETIME]" timezone="[TIMEZONE]">
     <uri-template>[URI TEMPLATE]uri-template>
     <done-flag>[FILE NAME]done-flag>
dataset>CopyErrorSuccess

舉例如下：

<dataset name="stats_hive_table" frequency="${coord:days(1)}" initial-instance="2016-12-25T00:00Z" timezone="America/Los_Angeles">
     <uri-template>
          hdfs://m1:9000/hive/warehouse/user_events/${YEAR}${MONTH}/${DAY}/data
     uri-template>
     <done-flag>donefile.flagdone-flag>
dataset>CopyErrorSuccess

上面會(huì)每天都會(huì)生成一個(gè)用戶事件表，可以供Hive查詢分析，這里指定了這個(gè)數(shù)據(jù)集的位置，后續(xù)計(jì)算會(huì)使用這部分?jǐn)?shù)據(jù)。其中，uri-template指定了一個(gè)匹配的模板，滿足這個(gè)模板的路徑都會(huì)被作為計(jì)算的基礎(chǔ)數(shù)據(jù)。 另外，還有一種定義dataset集合的方式，將多個(gè)dataset合并成一個(gè)組來(lái)定義，語(yǔ)法格式如下所示：

<datasets>
     <include>[SHARED_DATASETS]include>
     ...
     <dataset name="[NAME]" frequency="[FREQUENCY]" initial-instance="[DATETIME]" timezone="[TIMEZONE]">
          <uri-template>[URI TEMPLATE]uri-template>
     dataset>
     ...
datasets>CopyErrorSuccess

input-events和output-events元素

一個(gè)Coordinator Application的輸入事件指定了要執(zhí)行一個(gè)Coordinator動(dòng)作必須滿足的輸入條件，在Oozie當(dāng)前版本，只支持使用dataset實(shí)例。

一個(gè)Coordinator Action可能會(huì)生成一個(gè)或多個(gè)dataset實(shí)例，在Oozie當(dāng)前版本，輸出事件只支持輸出dataset實(shí)例。

常量表示形式含義說(shuō)明${coord:minutes(intn)}返回日期時(shí)間：從一開(kāi)始，周期執(zhí)行n分鐘${coord:hours(intn)}返回日期時(shí)間：從一開(kāi)始，周期執(zhí)行n*60分鐘${coord:days(intn)}返回日期時(shí)間：從一開(kāi)始，周期執(zhí)行n*24*60分鐘${coord:months(intn)}返回日期時(shí)間：從一開(kāi)始，周期執(zhí)行n*M*24*60分鐘（M表示一個(gè)月的天數(shù)）${coord:endOfDays(intn)}返回日期時(shí)間：從當(dāng)天的最晚時(shí)間（即下一天）開(kāi)始，周期執(zhí)行n*24*60分鐘${coord:endOfMonths(1)}返回日期時(shí)間：從當(dāng)月的最晚時(shí)間開(kāi)始（即下個(gè)月初），周期執(zhí)行n*24*60分鐘${coord:current(intn)}返回日期時(shí)間：從一個(gè)Coordinator動(dòng)作（Action）創(chuàng)建時(shí)開(kāi)始計(jì)算，第n個(gè)dataset實(shí)例執(zhí)行時(shí)間${coord:dataIn(Stringname)}在輸入事件（input-events）中，解析dataset實(shí)例包含的所有的URI${coord:dataOut(Stringname)}在輸出事件（output-events）中，解析dataset實(shí)例包含的所有的URI${coord:offset(intn,StringtimeUnit)}表示時(shí)間偏移，如果一個(gè)Coordinator動(dòng)作創(chuàng)建時(shí)間為T(mén)，n為正數(shù)表示向時(shí)刻T之后偏移，n為負(fù)數(shù)向向時(shí)刻T之前偏移，timeUnit表示時(shí)間單位（選項(xiàng)有MINUTE、HOUR、DAY、MONTH、YEAR）${coord:hoursInDay(intn)}指定的第n天的小時(shí)數(shù)，n>0表示向后數(shù)第n天的小時(shí)數(shù)，n=0表示當(dāng)天小時(shí)數(shù)，n<0表示向前數(shù)第n天的小時(shí)數(shù)${coord:daysInMonth(intn)}指定的第n個(gè)月的天數(shù)，n>0表示向后數(shù)第n個(gè)月的天數(shù)，n=0表示當(dāng)月的天數(shù)，n<0表示向前數(shù)第n個(gè)月的天數(shù)${coord:tzOffset()}dataset對(duì)應(yīng)的時(shí)區(qū)與CoordinatorJob的時(shí)區(qū)所差的分鐘數(shù)${coord:latest(intn)}最近以來(lái)，當(dāng)前可以用的第n個(gè)dataset實(shí)例${coord:future(intn,intlimit)}當(dāng)前時(shí)間之后的dataset實(shí)例，n>=0，當(dāng)n=0時(shí)表示立即可用的dataset實(shí)例，limit表示dataset實(shí)例的個(gè)數(shù)${coord:nominalTime()}nominal時(shí)間等于CoordinatorJob啟動(dòng)時(shí)間，加上多個(gè)CoordinatorJob的頻率所得到的日期時(shí)間。例如：start=”2009-01-01T24:00Z”，end=”2009-12-31T24:00Z”，frequency=”${coord:days(1)}”，frequency=”${coord:days(1)}，則nominal時(shí)間為：2009-01-02T00:00Z、2009-01-03T00:00Z、2009-01-04T00:00Z、…、2010-01-01T00:00Z${coord:actualTime()}Coordinator動(dòng)作的實(shí)際創(chuàng)建時(shí)間。例如：start=”2011-05-01T24:00Z”，end=”2011-12-31T24:00Z”，frequency=”${coord:days(1)}”，則實(shí)際時(shí)間為：2011-05-01，2011-05-02，2011-05-03，…，2011-12-31${coord:user()}啟動(dòng)當(dāng)前CoordinatorJob的用戶名稱${coord:dateOffset(StringbaseDate,intinstance,StringtimeUnit)}計(jì)算新的日期時(shí)間的公式：newDate=baseDate+instance*timeUnit，如：baseDate=’2009-01-01T00:00Z’，instance=’2′，timeUnit=’MONTH’，則計(jì)算得到的新的日期時(shí)間為’2009-03-01T00:00Z’。${coord:formatTime(StringtimeStamp,Stringformat)}格式化時(shí)間字符串，format指定模式。