在Hadoop中執行的任務有時候需要把多個Map/Reduce作業連接到一起，這樣才能夠達到目的。[1]在Hadoop生態圈中，有一種相對比較新 的組件叫做Oozie[2]，它讓我們可以把多個Map/Reduce作業組合到一個邏輯工作單元中，從而完成更大型的任務。本文中，我們會向你介紹 Oozie以及使用它的一些方式。
什么是Oozie？
Oozie是一種Java Web應用程序，它運行在Java servlet容器——即Tomcat——中，並使用數據庫來存儲以下內容：
- 工作流定義
- 當前運行的工作流實例，包括實例的狀態和變量
Oozie工作流是放置在控制依賴DAG（有向無環圖 Direct Acyclic Graph）中的一組動作（例如，Hadoop的Map/Reduce作業、Pig作業等），其中指定了動作執行的順序。我們會使用hPDL（一種XML流程定義語言）來描述這個圖。
hPDL是一種很簡潔的語言，只會使用少數流程控制和動作節點。控制節點會定義執行的流程，並包含工作流的起點和終點（start、end和 fail節點）以及控制工作流執行路徑的機制（decision、fork和join節點）。動作節點是一些機制，通過它們工作流會觸發執行計算或者處理 任務。Oozie為以下類型的動作提供支持： Hadoop map-reduce、Hadoop文件系統、Pig、Java和Oozie的子工作流（SSH動作已經從Oozie schema 0.2之后的版本中移除了）。
所有由動作節點觸發的計算和處理任務都不在Oozie之中——它們是由Hadoop的Map/Reduce框架執行的。這種方法讓Oozie可以支 持現存的Hadoop用於負載平衡、災難恢復的機制。這些任務主要是異步執行的（只有文件系統動作例外，它是同步處理的）。這意味着對於大多數工作流動作 觸發的計算或處理任務的類型來說，在工作流操作轉換到工作流的下一個節點之前都需要等待，直到計算或處理任務結束了之后才能夠繼續。Oozie可以通過兩 種不同的方式來檢測計算或處理任務是否完成，也就是回調和輪詢。當Oozie啟動了計算或處理任務的時候，它會為任務提供唯一的回調URL，然后任務會在 完成的時候發送通知給特定的URL。在任務無法觸發回調URL的情況下（可能是因為任何原因，比方說網絡閃斷），或者當任務的類型無法在完成時觸發回調 URL的時候，Oozie有一種機制，可以對計算或處理任務進行輪詢，從而保證能夠完成任務。
Oozie工作流可以參數化（在工作流定義中使用像${inputDir}之類的變量）。在提交工作流操作的時候，我們必須提供參數值。如果經過合適地參數化（比方說，使用不同的輸出目錄），那么多個同樣的工作流操作可以並發。
一些工作流是根據需要觸發的，但是大多數情況下，我們有必要基於一定的時間段和（或）數據可用性和（或）外部事件來運行它們。Oozie協調系統 （Coordinator system）讓用戶可以基於這些參數來定義工作流執行計划。Oozie協調程序讓我們可以以謂詞的方式對工作流執行觸發器進行建模，那可以指向數據、事 件和（或）外部事件。工作流作業會在謂詞得到滿足的時候啟動。
經常我們還需要連接定時運行、但時間間隔不同的工作流操作。多個隨后運行的工作流的輸出會成為下一個工作流的輸入。把這些工作流連接在一起，會讓系統把它作為數據應用的管道來引用。Oozie協調程序支持創建這樣的數據應用管道。
安裝Oozie
我們可以把Oozie安裝在現存的Hadoop系統中，安裝方式包括tarball、RPM和Debian包等。我們的Hadoop部署是Cloudera的CDH3，其中已經包含了Oozie。因此，我們只是使用yum把它拉下來，然后在edge節點[1]上 執行安裝操作。在Oozie的發布包中有兩個組件——Oozie-client和Oozie-server。根據簇集的規模，你可以讓這兩個組件安裝在同 一台edge服務器上，也可能安裝在不同的計算機上。Oozie服務器中包含了用於觸發和控制作業的組件，而客戶端中包含了讓用戶可以觸發Oozie操作 並與Oozie服務器通信的組件。
想要了解更多關於安裝過程的信息，請使用Cloudera發布包，並訪問Cloudera站點[2]。
注： 除了包括安裝過程的內容之外，它還建議把下面的shell變量OOZIE_URL根據需要添加到.login、.kshrc或者shell的啟動文件中：
(export OOZIE_URL=http://localhost:11000/oozie)
簡單示例
為了向你展示Oozie的使用方法，讓我們創建一個簡單的示例。我們擁有兩個Map/Reduce作業[3]——一個會獲取最初的數據，另一個會合並指定類型的數據。實際的獲取操作需要執行最初的獲取操作，然后把兩種類型的數據——Lidar和Multicam——合並。為了讓這個過程自動化，我們需要創建一個簡單的Oozie工作流（代碼1）。
<!--
Copyright (c) 2011 NAVTEQ! Inc. All rights reserved.
NGMB IPS ingestor Oozie Script
-->
<workflow-app xmlns='uri:oozie:workflow:0.1' name='NGMB-IPS-ingestion'>
<start to='ingestor'/>
<action name='ingestor'>
<java>
<job-tracker>${jobTracker}</job-tracker>
<name-node>${nameNode}</name-node>
<configuration>
<property>
<name>mapred.job.queue.name</name>
<value>default</value>
</property>
</configuration>
<main-class>com.navteq.assetmgmt.MapReduce.ips.IPSLoader</main-class>
<java-opts>-Xmx2048m</java-opts>
<arg>${driveID}</arg>
</java>
<ok to="merging"/>
<error to="fail"/>
</action>
<fork name="merging">
<path start="mergeLidar"/>
<path start="mergeSignage"/>
</fork>
<action name='mergeLidar'>
<java>
<job-tracker>${jobTracker}</job-tracker>
<name-node>${nameNode}</name-node>
<configuration>
<property>
<name>mapred.job.queue.name</name>
<value>default</value>
</property>
</configuration>
<main-class>com.navteq.assetmgmt.hdfs.merge.MergerLoader</main-class>
<java-opts>-Xmx2048m</java-opts>
<arg>-drive</arg>
<arg>${driveID}</arg>
<arg>-type</arg>
<arg>Lidar</arg>
<arg>-chunk</arg>
<arg>${lidarChunk}</arg>
</java>
<ok to="completed"/>
<error to="fail"/>
</action>
<action name='mergeSignage'>
<java>
<job-tracker>${jobTracker}</job-tracker>
<name-node>${nameNode}</name-node>
<configuration>
<property>
<name>mapred.job.queue.name</name>
<value>default</value>
</property>
</configuration>
<main-class>com.navteq.assetmgmt.hdfs.merge.MergerLoader</main-class>
<java-opts>-Xmx2048m</java-opts>
<arg>-drive</arg>
<arg>${driveID}</arg>
<arg>-type</arg>
<arg>MultiCam</arg>
<arg>-chunk</arg>
<arg>${signageChunk}</arg>
</java>
<ok to="completed"/>
<error to="fail"/>
</action>
<join name="completed" to="end"/>
<kill name="fail">
<message>Java failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
</kill>
<end name='end'/>
</workflow-app>
代碼1： 簡單的Oozie工作流
這個工作流定義了三個動作：ingestor、mergeLidar和mergeSignage。並把每個動作都實現為Map/Reduce[4]作業。這個工作流從start節點開始，然后把控制權交給Ingestor動作。一旦ingestor步驟完成，就會觸發fork控制節點 [4]，它會並行地開始執行mergeLidar和mergeSignage[5]。這兩個動作完成之后，就會觸發join控制節點[6]。join節點成功完成之后，控制權就會傳遞給end節點，它會結束這個過程。
創建工作流之后，我們需要正確地對其進行部署。典型的Oozie部署是一個HDFS目錄，其中包含workflow.xml（代碼1）、config-default.xml和lib子目錄，其中包含有工作流操作所要使用的類的jar文件。
（點擊可以查看大圖）
圖1： Oozie部署
config-default.xml文件是可選的，通常其中會包含對於所有工作流實例通用的工作流參數。代碼2中顯示的是config-default.xml的簡單示例。
<configuration>
<property>
<name>jobTracker</name>
<value>sachicn003:2010</value>
</property>
<property>
<name>nameNode</name>
<value>hdfs://sachicn001:8020</value>
</property>
<property>
<name>queueName</name>
<value>default</value>
</property>
</configuration>
代碼2： Config-default.xml
完成了工作流的部署之后，我們可以使用Oozie提供的命令行工具[5]，它可以用於提交、啟動和操作工作流。這個工具一般會運行在Hadoop簇集[7]的edge節點上，並需要一個作業屬性文件（參見配置工作流屬性），見代碼3。
oozie.wf.application.path=hdfs://sachicn001:8020/user/blublins/workflows/IPSIngestion jobTracker=sachicn003:2010 nameNode=hdfs://sachicn001:8020
代碼3： 作業屬性文件
有了作業屬性，我們就可以使用代碼4中的命令來運行Oozie工作流。
oozie job –oozie http://sachidn002.hq.navteq.com:11000/oozie/ -D driveID=729-pp00002-2011-02-08-09-59-34 -D lidarChunk=4 -D signageChunk=20 -config job.properties –run
列表4： 運行工作流命令
配置工作流屬性在config-default.xml、作業屬性文件和作業參數中有一些重疊，它們可以作為命令行調用的一部分傳遞給Oozie。盡管文檔中沒有清晰地指出何時使用哪個，但總體上的建議如下：
Oozie處理這三種參數的方式如下：
|
我們可以使用Oozie控制台（圖2）來觀察工作流執行的進程和結果。
（點擊可以查看大圖）
圖2： Oozie控制台
我們還可以使用Oozie控制台來獲得操作執行的細節，比方說作業的日志[8]（圖3）。
（點擊可以查看大圖）
圖3： Oozie控制台——作業日志
編程方式的工作流調用
盡管上面所述的命令行界面能夠很好地用於手動調用Oozie，但有時使用編程的方式調用Oozie更具有優勢。當Oozie工作流是特定的應用程序 或者大型企業過程的一部分，這就會很有用。我們可以使用Oozie Web Services APIs [6]或者Oozie Java client APIs [7]來實現這種編程方式的調用。代碼5中展現的就是很簡單的Oozie Java客戶端的例子，它會觸發上面描述的過程。
package com.navteq.assetmgmt.oozie;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.client.OozieClientException;
import org.apache.oozie.client.WorkflowJob;
import org.apache.oozie.client.WorkflowJob.Status;
public class WorkflowClient {
private static String OOZIE_URL = "http://sachidn002.hq.navteq.com:11000/oozie/";
private static String JOB_PATH = "hdfs://sachicn001:8020/user/blublins/workflows/IPSIngestion";
private static String JOB_Tracker = "sachicn003:2010";
private static String NAMENode = "hdfs://sachicn001:8020";
OozieClient wc = null;
public WorkflowClient(String url){
wc = new OozieClient(url);
}
public String startJob(String wfDefinition, List<WorkflowParameter> wfParameters)
throws OozieClientException{
// create a workflow job configuration and set the workflow application path
Properties conf = wc.createConfiguration();
conf.setProperty(OozieClient.APP_PATH, wfDefinition);
// setting workflow parameters
conf.setProperty("jobTracker", JOB_Tracker);
conf.setProperty("nameNode", NAMENode);
if((wfParameters != null) && (wfParameters.size() > 0)){
for(WorkflowParameter parameter : wfParameters)
conf.setProperty(parameter.getName(), parameter.getValue());
}
// submit and start the workflow job
return wc.run(conf);
}
public Status getJobStatus(String jobID) throws OozieClientException{
WorkflowJob job = wc.getJobInfo(jobID);
return job.getStatus();
}
public static void main(String[] args) throws OozieClientException, InterruptedException{
// Create client
WorkflowClient client = new WorkflowClient(OOZIE_URL);
// Create parameters
List<WorkflowParameter> wfParameters = new LinkedList<WorkflowParameter>();
WorkflowParameter drive = new WorkflowParameter("driveID","729-pp00004-2010-09-01-09-46");
WorkflowParameter lidar = new WorkflowParameter("lidarChunk","4");
WorkflowParameter signage = new WorkflowParameter("signageChunk","4");
wfParameters.add(drive);
wfParameters.add(lidar);
wfParameters.add(signage);
// Start Oozing
String jobId = client.startJob(JOB_PATH, wfParameters);
Status status = client.getJobStatus(jobId);
if(status == Status.RUNNING)
System.out.println("Workflow job running");
else
System.out.println("Problem starting Workflow job");
}
}
代碼5： 簡單的Oozie Java客戶端
在此，我們首先使用Oozie服務器URL對工作流客戶端進行初始化。初始化過程完成之后，我們就可以使用客戶端提交並啟動作業（startJob方法），獲得正在運行的作業的狀態（getStatus方法），以及進行其他操作。
構建java動作，向工作流傳遞參數
在之前的示例中，我們已經展示了如何使用標簽向Java節點傳遞參數。由於Java節點是向Oozie引入自定義計算的主要方法，因此能夠從Java節點向Oozie傳遞數據也同樣重要。
根據Java節點的文檔[3]，我們可以使用“capture-output””元素把Java節點生成的值傳遞回給Oozie 上下文。然后，工作流的其它步驟可以通過EL-functions訪問這些值。返回值需要以Java屬性格式文件寫出來。我們可以通過 “JavaMainMapper.OOZIE_JAVA_MAIN_CAPTURE_OUTPUT_FILE”常量從System屬性中獲得這些屬性文件 的名稱。代碼6是一個簡單示例，演示了如何完成這項操作。
package com.navteq.oozie;
import java.io.File;
import java.io.FileOutputStream;
import java.io.OutputStream;
import java.util.Calendar;
import java.util.GregorianCalendar;
import java.util.Properties;
public class GenerateLookupDirs {
/**
* @param args
*/
public static final long dayMillis = 1000 * 60 * 60 * 24;
private static final String OOZIE_ACTION_OUTPUT_PROPERTIES = "oozie.action.output.properties";
public static void main(String[] args) throws Exception {
Calendar curDate = new GregorianCalendar();
int year, month, date;
String propKey, propVal;
String oozieProp = System.getProperty(OOZIE_ACTION_OUTPUT_PROPERTIES);
if (oozieProp != null) {
File propFile = new File(oozieProp);
Properties props = new Properties();
for (int i = 0; I < 8; ++i) {
year = curDate.get(Calendar.YEAR);
month = curDate.get(Calendar.MONTH) + 1;
date = curDate.get(Calendar.DATE);
propKey = "dir"+i;
propVal = year + "-" +
(month < 10 ? "0" + month : month) + "-" +
(date < 10 ? "0" + date : date);
props.setProperty(propKey, propVal);
curDate.setTimeInMillis(curDate.getTimeInMillis() - dayMillis);
}
OutputStream os = new FileOutputStream(propFile);
props.store(os, "");
os.close();
} else
throw new RuntimeException(OOZIE_ACTION_OUTPUT_PROPERTIES
+ " System property not defined");
}
}
代碼6： 向Oozie傳遞參數
在這個示例中，我們假設在HDFS中有針對每個日期的目錄。這樣，這個類首先會獲得當前日期，然后再獲得離現在最近的7個日期（包括今天），然后把目錄名稱傳遞回給Oozie。
結論
在本文我們介紹了Oozie，它是針對Hadoop的工作流引擎，並且提供了使用它的簡單示例。在下一篇文章中，我們會看到更復雜的例子，讓我們可以更進一步討論Oozie的特性。
致謝
非常感謝我們在Navteq的同事Gregory Titievsky，他為我們提供了一些例子。
Boris Lublinsky是NAVTEQ公司的首席架構師，在這家公司中他的工作是為大型數據管理和處理、SOA以及實現各種 NAVTEQ的項目定義架構的願景。 他還是InfoQ的SOA編輯，以及OASIS的SOA RA工作組的參與者。Boris是一位作者，還經常發表演講，他最新的一本書是《Applied SOA》。
Michael Segel在過去二十多年間一直與客戶寫作，識別並解決他們的業務問題。 Michael已經作為多種角色、在多個行業中工作過。他是一位獨立顧問，總是期望能夠解決所有有挑戰的問題。Michael擁有俄亥俄州立大學的軟件工程學位。
[1]edge節點是安裝有Hadoop庫的計算機，但不是真正簇集中的一部分。它是為能夠連接到簇集中的應用程序所用的，並且會部署輔助服務以及能夠直接訪問簇集的最終用戶應用程序。
[3]這些作業的細節和本文無關，所以在其中沒有描述。
[4]Map/Reduce 作業能夠以兩種不同的方式在Oozie中實現——第一種是作為真正的Map/Reduce動作[2]，其中你會指定Mapper和Reducer類以及它 們的配置信息；第二種是作為Java動作[3]，其中你會使用Hadoop API來指定啟動Map/Reduce作業的類。因為我們所有的Java主函數都是使用Hadoop API，並且還實現了一些額外的功能，所以我們選擇了第二種方法。
[5] Oozie確保兩個動作會並行地提交給作業跟蹤程序。在執行過程中實際的並行機制並不在Oozie的控制之內，並且依賴於作業的需求、簇集的能力以及Map/Reduce部署所使用的調度程序。
[6]join動作的功能是要同步fork動作啟動的多個並行執行的線程。如果fork啟動的所有執行的線程都能夠成功完成，那么join動作就會等待它們全部完成。如果有至少一個線程執行失敗，kill節點會“殺掉”剩余運行的線程。
[7] 這個節點不需要是安裝了Oozie的計算機。
[8] Oozie的作業日志會包含工作流執行的細節，想要查看動作執行的細節，我們需要切換到Hadoop的Map/Reduce管理頁面。
查看英文原文：Introduction to Oozie