工作流引擎Oozie(二):coordinator


1. 簡介

coordinator是workflow的定時提交器,基於時間條件與數據生成觸發(based on time and data triggers)。簡單點說,coordinator按所定義的時間周期進行輪詢,若數據生成條件滿足,則觸發workflow任務;否則,則等待數據生成或跳過(調度策略由設置的超時時間決定)。

2. 詳解

coordinator-app

<coordinator-app name="coord-demo" frequency="${coord:days(1)}" start="${start_time}" end="${end_time}"
                 timezone="Asia/Shanghai" xmlns="uri:oozie:coordinator:0.1">

文件頭定義了coordinator任務的名稱,執行頻率(frequency),開始與起止時間(start、end)。${coord:days(1)}為coordinator內置的EL function (Expression Language),表示為執行周期為一天。frequency的時間單位為分鍾,則${coord:days(1)}=1440${coord:hours(3)}=180. 一般地,在job.properties定義如:start_time=2016-06-07T00:00Z,應以UTC時區為准。

Control information

<controls>
	<!--超時時間,若為0,一旦數據生成超時則coordinator action被丟棄;
	若為-1,則一直等待數據生成;
	若 > 0,即為超時時間(單位:分鍾)-->
	<timeout>-1</timeout>
	<!--並發action數-->
	<concurrency>1</concurrency>
	<!--同時被觸發的多個coordinator job的執行次序-->
	<execution>${execution_order}</execution>
</controls>

datasets

<datasets>
    <!--數據生成目錄-->
    <dataset name="data" frequency="${coord:hours(1)}" initial-instance="${start_time}" timezone="${timezone}">
      <uri-template>${nameNode}/path/to/${YEAR}-${MONTH}-${DAY}/${HOUR}/</uri-template>
      <done-flag></done-flag>
    </dataset>
</datasets>

相關參數:

  • initial-instance,對應初始時間的數據目錄,作為時間周期偏移的基准目錄
  • uri-template,文件目錄HDFS目錄,${YEAR}${MONTH}等為EL Time Constants,具體數值由<input-events> <data-in ..>傳入,也可通過job.properties傳入自定義的參數。
  • done-flag,數據目錄生成的文件標識,若未指定,則默認為 _SUCCESS 文件;若指定為空,則表示文件夾本身。

input-events

<input-events>
	<data-in name="dataReady" dataset="data">
	  <instance>${coord:current(-1)}</instance>
	</data-in>
</input-events>

定義了數據觸發條件,${coord:current(-1)}表示對於當前時間的偏移,其中偏移量=數值* dataset frequency。有些時候,我們要判斷多個目錄是否生成,比如,判斷一天的24個小時目錄,這時應這樣寫:

<data-in name="dataReady2" dataset="data">
  <start-instance>${coord:current(-24)}</start-instance>
  <end-instance>${coord:current(-1)}</end-instance>
</data-in>

action

<action>
	<workflow>
	  <app-path>${wf_app_path}</app-path>
	  <configuration>
	    <property>
	      <name>dayTime</name>
	      <value>${coord:formatTime(coord:dateOffset(coord:nominalTime(), -1, 'DAY'), 'yyyy-MM-dd')}</value>
	    </property>
	  </configuration>
	</workflow>
</action>

定義coordinator要觸發的workflow,

  • app-path,workflow所在的HDFS目錄;
  • property,定義傳給workflow的參數

coordinator提供了一些用於時間計算的函數,比如:

  • coord:nominalTime(),返回coordinator的每一執行周期的當前時間;
  • coord:dateOffset(String baseDate, int instance, String timeUnit),計算偏移后的時間,newDate = baseDate + instance * timeUnit;
  • coord:formatTime(String ts, String format),定義格式化的時間,其中format應遵循Java's SimpleDateFormat。

返回多個時間:

<value>${coord:formatTime(coord:dateOffset(coord:dateOffset(coord:nominalTime(), -12, 'HOUR'), -6, 'DAY'), 'yyyy-MM-dd')},${coord:formatTime(coord:dateOffset(coord:dateOffset(coord:nominalTime(), -12, 'HOUR'), -5, 'DAY')</value>


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM