在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流程定義語言）來描述這個圖。
相關廠商内容
看Scala如何颠覆傳統金融企業
高可用網貸系統讓您投資無憂
微衆架構首秀--基于自主可控技術的分布式架構實踐
企業級SaaS應用平台--釘釘技術曆程之路
相關贊助商
全球架構師峰會，12月18-19日，北京·國際會議中心，9折報名截止11月27日！
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控制台（圖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庫的計算機，但不是真正簇集中的一部分。它是為能夠連接配接到簇集中的應用程式所用的，并且會部署輔助服務以及能夠直接通路簇集的最終使用者應用程式。
[2]請參看Oozie安裝的連結。
[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