Hadoop 的 Oozie 工作流管理引擎的實際應用（一）

阿新 • • 發佈：2019-02-08

在 IBM Bluemix 雲平臺上開發並部署您的下一個應用。

簡介

Apache Oozie 是用於 Hadoop 平臺的一種工作流排程引擎。該框架（如圖 1 所示）使用 Oozie 協調器促進了相互依賴的重複工作之間的協調，您可以使用預定的時間或資料可用性來觸發 Apache Oozie。您可以使用 Oozie bundle 系統提交或維護一組協調應用程式。作為本練習的一部分，Oozie 運行了一個 Apache Sqoop 作業，以便在 MySQL 資料庫中的資料上執行匯入操作，並將資料傳輸到 Hadoop 分散式檔案系統 (HDFS) 中。可以利用匯入的資料集執行 Sqoop 合併操作，從而更新較舊的資料集。通過利用 UNIX shell 操作，可從 MySQL 資料庫中提取用來執行 Sqoop 作業的元資料。同理，可執行 Java 操作來更新 Sqoop 作業所需的 MySQL 資料庫中的元資料。

InfoSphere BigInsights Quick Start Edition

InfoSphere BigInsights Quick Start Edition 是一個免費的、可下載的 InfoSphere BigInsights 版本，是 IBM 基於 Hadoop 的產品。使用 Quick Start Edition，您可以嘗試使用 IBM 開發的特性來提高開源 Hadoop 的價值，這些特性包括 Big SQL、文字分析和 BigSheets。為了讓您的體驗儘可能順利，我們提供了引導式學習，包括一些按部就班的、自定進度的教程和視訊，它們可以幫助您開始讓 Hadoop 為您工作。沒有時間或資料的限制，您可以自行安排時間在大量資料上進行試驗。

觀看視訊、遵循這些教程 (PDF) 並立刻下載 BigInsights Quick Start Edition。

圖 1. Oozie 編排架構

回頁首

需要安裝的軟體

要想充分利用本文的示例，訪問以下軟體可能對您有所幫助：

該叢集是一個分散式叢集，在 1 個主名稱節點、2 個核心節點和 8 個任務節點中執行。

回頁首

Oozie 工作流

Oozie 工作流是控制依賴有向非迴圈圖 (DAG) 中安排的 Oozie 操作的集合。控制依賴（Control dependency）可確保以下操作在前面的操作已成功完成後才會啟動。本文首先會簡要概述工作流控制節點，然後重點介紹以下工作流操作節點：

回頁首

工作流控制節點

啟動控制節點（如清單 1 所示）是工作流作業的入口點。在工作流啟動時，它會自動過渡到啟動過程中指定的節點。

清單 1. 啟動控制節點

<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf">
    <start to="timeCheck"/>
</workflow-app>

末端控制節點（如清單 2 所示）是結束工作流作業時所用的節點。它表示工作流操作已經成功完成。一個工作流定義必須有一個末端節點。

清單 2. 末端控制節點

<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf">
    <end name="end"/>
</workflow-app>

停止控制節點（如清單 3 所示）可使工作流作業自行停止。在到達停止節點（kill node）時，如果工作流作業啟動的一個或更多操作正在執行，那麼當前執行的所有操作都將停止。工作流定義可以包含零個或更多停止節點。

清單 3. 停止控制節點

<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf">
    <kill name="fail">
        <message>Sqoop failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
   </kill>
</workflow-app>

決策控制節點（如清單 4 所示）可使工作流確定要採用的執行路徑。決策節點的工作原理類似於擁有一組謂詞轉換對（predicates-transition pair）和一個預設轉換的 switch-case 塊。謂詞是按順序進行評估的，直至其中一個評估為 ture 為止，同時還會進行相應的轉換。如果沒有一個謂詞被評估為 true，則會採用預設轉換。

清單 4. 決策控制節點

<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf">
    <decision name="master-decision">
       <switch>
         <case to="sqoopMerge1">
                 ${wf:actionData('hiveSwitch')['paramNum'] eq 1}
         </case>
         <default to="sqoopMerge2"/>
       </switch>
   </decision>
</workflow-app>

分支節點將一個執行路徑分為多個併發路徑。聯接節點一直等待，直到前面的分支節點的所有併發執行路徑都到達聯接節點為止。您必須成對使用分叉節點和聯接節點，如清單 5 所示。

清單 5. 分支-聯接控制節點

<workflow-app xmlns="uri:oozie:workflow:0.2" name="ooziedemo-wf">
    <fork name="forking">
        <path start="sqoopMerge1"/>
        <path start="sqoopMerge2"/>
    </fork>
    <join name="joining" to="hiveSwitch"/>
</workflow-app>

回頁首

Oozie shell 操作

您可以將 Oozie shell 操作作為工作流的一部分進行配置，從而執行某個檔案中的一組 shell 指令碼。您可以利用包含必要引數的 job-tracker、name-node 和 exec 元素來配置 Oozie shell 操作，從而執行相關的任務，如清單 6 所示。您可以將具有配置引數的一個配置 shell 操作來建立或刪除 HDFS 上的檔案和目錄，然後啟動 shell 作業。您可以使用與配置元素內聯的 job-xml 元素，利用配置引數將一個 XML 檔案傳遞給 shell 作業。您可以配置其他檔案或歸檔檔案，讓它們可用於 shell 作業。在 shell 作業結束後，您可以讓 shell 作業的輸出可供workflow 作業使用，但它需要滿足以下條件：

輸出的格式必須是一個有效的 Java 屬性檔案。
輸出的大小必須小於 2KB。

清單 6. Shell 指令碼

host="XXX.XX.XX.XXX"
port="3306"
username="root"
password=""
database="zzz"
tableName="$1"

####################################
echo "Host: $host"
echo "Database: $database"
echo "Table: $tableName"
####################################

sqoopLstUpd=`mysql --host=$host --port=$port --user=$username --password=$password 
-N -e 'SELECT PARM_DATE_VAL from T_CONTROL_PARM where PARM_NM="SQOOP_INCR_LST_UPD"
 and PARM_GROUP_NM="'$tableName'"' $database`

echo "sqoopLstUpd=$sqoopLstUpd"
echo "tableName=$tableName"

清單 7 展示了 workflow.xml 檔案中的 shell 操作配置。

清單 7. Oozie shell 操作

<action name="timeCheck">
    <shell xmlns="uri:oozie:shell-action:0.1">
       <job-tracker>${jobTracker}</job-tracker>
       <name-node>${nameNode}</name-node>
       <configuration>
           <property>
               <name>mapred.job.queue.name</name>
               <value>${queueName}</value>
           </property>
       </configuration>
       <exec>${sqoopUpdTrack}</exec>
       <argument>${tableName}</argument>
       <file>${sqoopUpdTrackPath}#${sqoopUpdTrack}</file>
       <capture-output/>
    </shell>
    <ok to="sqoopIncrImport"/>
    <error to="fail"/>
</action>

要想訪問 shell 輸出，可以使用清單 8 中所示的 Sqoop 增量作業。

清單 8. 用來實現增量匯入的 Oozie Sqoop 操作

<action name="sqoopIncrImport">
        <sqoop xmlns="uri:oozie:sqoop-action:0.2">
           <job-tracker>${jobTracker}</job-tracker>
           <name-node>${nameNode}</name-node>
           <prepare>
               <delete path="${s3BucketLoc}/${tableName}/incr"/>
               <mkdir path="${s3BucketLoc}/${tableName}"/>
           </prepare>
           <configuration>
               <property>
                   <name>mapred.job.queue.name</name>
                   <value>${queueName}</value>
               </property>
           </configuration>
           <arg>import</arg>
           <arg>--connect</arg>
           <arg>${dbURL}</arg>
           <arg>--driver</arg>
           <arg>${mySqlDriver}</arg>
           <arg>--username</arg>
           <arg>${user}</arg>
           <arg>--table</arg>
           <arg>${wf:actionData('timeCheck')['tableName']}</arg>
           <arg>--target-dir</arg>
           <arg>${s3BucketLoc}/${tableName}/incr</arg>
           <arg>--check-column</arg>
           <arg>LAST_UPD</arg>
           <arg>--incremental</arg>
           <arg>lastmodified</arg>
           <arg>--last-value</arg>
           <arg>${wf:actionData('timeCheck')['sqoopLstUpd']}</arg>
           <arg>--m</arg>
           <arg>1</arg>
       </sqoop>
       <ok to="sqoopMetaUpdate"/>
       <error to="fail"/>
   </action>

回頁首

Oozie Java 操作

Java 操作運行了指定的主要 Java 類的 public static void main (String [] args) 方法。Java 應用程式作為具有單個 mapper 任務的 MapReduce 作業執行在 Hadoop 叢集上。工作流作業一直要等到 Java 操作結束執行之後才能繼續執行下一個操作。Java 操作可使用 job-tracker、name-node、Java 主類、JVM 選項和輸入引數進行配置，如清單 9 所示。您可以使用 Expression Language (EL) 表示式將引數分配給內聯屬性值。您必須以 Java 屬性檔案的格式寫入所有輸出引數。

您可以配置 Java 操作來清理 HDFS 檔案和目錄，或者建立 Apache HCatalog 分割槽，然後再啟動 Java 應用程式。這使得 Oozie 能夠在出現暫時性或非暫時性故障時重試 Java 操作。

清單 9. Oozie Java 操作

<action name="sqoopMetaUpdate">
          <java>
               <job-tracker>${jobTracker}</job-tracker>
               <name-node>${nameNode}</name-node>
               <configuration>
                   <property>
                      <name>mapred.job.queue.name</name>
                      <value>${queueName}</value>
                   </property>
               </configuration>
               <main-class>SqoopMetaUtil</main-class>
               <java-opts></java-opts>
               <arg>${tableName}</arg>
               <archive>${mySqlDriverPath}</archive>
          </java>
          <ok to="hiveSwitch"/>
          <error to="fail"/>
</action>

您可以通過使用 capture-output 來配置 Java 操作，從而將值傳遞給下一操作。您可以通過使用 Hadoop EL 函式來訪問這些值。您能夠以 Java 屬性檔案的格式在 Java 類中寫入值，如清單 10 所示。

清單 10. 用於實現值傳遞的 Java 程式碼片段

String OOZIE_ACTION_OUTPUT_PROPERTIES = "oozie.action.output.properties";
String oozieProp = System.getProperty(OOZIE_ACTION_OUTPUT_PROPERTIES);
       OutputStream os = null;
       if(oozieProp != null){
          File propFile = new File(oozieProp);
          Properties p = new Properties();
          p.setProperty("name", "Autodesk");
          p.setProperty("address", "Sun Rafael");
          try {
               os = new FileOutputStream(propFile);
               p.store(os, "");
          } catch (FileNotFoundException e) {
               System.err.println("<<< FileNotFoundException >>>"+e.getMessage());
          } catch (IOException e) {
               System.err.println("<<< IOException >>>"+e.getMessage());
          }
          finally{
               if(os != null)
               try {
                    os.close();
               } catch (IOException e) {
                    System.err.println("<<< IOException >>>"+e.getMessage());
               }
          }
       }
       else{
            throw new RuntimeException(OOZIE_ACTION_OUTPUT_PROPERTIES
                    + " System property not defined");
    }

您可以在 workflow.xml 檔案中配置操作來訪問屬性檔案中的相應值設定，如清單 11 所示。

清單 11. 用於實現值傳遞的 Oozie Java 操作

<action name="jProperties">
      <java>
           <job-tracker>${jobTracker}</job-tracker>
           <name-node>${nameNode}</name-node>
           <configuration>
               <property>
                  <name>mapred.job.queue.name</name>
                  <value>${queueName}</value>
               </property>
           </configuration>
           <main-class>PropertyExplorer</main-class>
           <java-opts></java-opts>
           <capture-output/>
      </java>
      <ok to="email"/>
      <error to="fail"/>
   </action>

   <action name="email">
         <email xmlns="uri:oozie:email-action:0.1">
            <to>[email protected]</to>
            <subject>Oozie workflow finished successfully!</subject>
            <body>${wf:actionData('jProperties')['name']} | 
            ${wf:actionData('jProperties')['address']}</body>
         </email>
         <ok to="end"/>
         <error to="fail"/>
   </action>

回頁首

Oozie Sqoop 操作

Oozie 工作流觸發了一個 Sqoop 指令碼，該指令碼在 Hadoop 叢集上啟動了一個 Sqoop 作業。Sqoop 作業通過在 Hadoop 叢集上啟動 MapReduce 作業來完成任務。Sqoop 指令碼啟動的 MapReduce 作業會將資料從 RDBMS 傳輸到 HDFS。您可以配置一個 Sqoop 操作（如清單 12 所示）來刪除 HDFS 上的檔案和目錄，然後再啟動 Sqoop 作業。與其他 Oozie 操作類似，您可以通過使用 job-xml 元素，利用其他的屬性來配置 Sqoop 操作。configuration 元素中指定的屬性值將會覆蓋 job-xml 元素中指定的屬性。可以將其他檔案和歸檔檔案提供給 Sqoop 作業。

清單 12. 用於合併的 Oozie Sqoop 操作

<action name="sqoopMerge1">
         <sqoop xmlns="uri:oozie:sqoop-action:0.2">
             <job-tracker>${jobTracker}</job-tracker>
             <name-node>${nameNode}</name-node>
             <prepare>
                 <delete path="${s3BucketLoc}/${tableName}/master1"/>
                 <mkdir path="${s3BucketLoc}/${tableName}"/>
             </prepare>
             <configuration>
                 <property>
                     <name>mapred.job.queue.name</name>
                     <value>${queueName}</value>
                 </property>
             </configuration>
             <arg>merge</arg>
             <arg>--new-data</arg>
             <arg>${s3incr}</arg>
             <arg>--onto</arg>
             <arg>${s3BucketLoc}/${tableName}/master2</arg>
             <arg>--target-dir</arg>
             <arg>${s3BucketLoc}/${tableName}/master1</arg>
             <arg>--jar-file</arg>
             <arg>${tableJarLoc}/${tableName}.jar</arg>
             <arg>--class-name</arg>
             <arg>${tableName}</arg>
             <arg>--merge-key</arg>
             <arg>ROW_ID</arg>
         </sqoop>
         <ok to="hive-master1"/>
         <error to="fail"/>
   </action>

回頁首

Oozie Hive 操作

您可以配置 Hive 操作（如清單 13 所示）來執行 HDFS 上的檔案和目錄中的任何 Hive 指令碼。該操作啟動了一個 MapReduce 作業來完成這些任務。您需要在 Oozie 上配置 Hive 操作，使用 Hive 配置檔案 hive-default.xml 或 hive-site.xml 作為 job-xml 元素。對於支援 Hive 訪問 Oozie 環境而言，這一點是必需的。您可以配置 Hive 操作來建立或刪除 HDFS 檔案和目錄，然後再啟動 Hive 作業。configuration 元素中指定的屬性值將會覆蓋 job-xml 檔案中指定的值。您可以新增其他的檔案和歸檔檔案，讓它們可用於 Hive 作業。Oozie 執行了由指令碼元素中的路徑指定的 Hive 指令碼。您可以通過 Oozie 工作流，將引數作為輸入引數分配給 Hive 指令碼。

清單 13. Oozie Hive 操作

<action name="hiveSwitch">
     <shell xmlns="uri:oozie:shell-action:0.1">
         <job-tracker>${jobTracker}</job-tracker>
         <name-node>${nameNode}</name-node>
          <configuration>
          <property>
              <name>mapred.job.queue.name</name>
              <value>${queueName}</value>
          </property>
          </configuration>
          <exec>${hiveSwitchScript}</exec>
          <argument>${tableName}</argument>
          <file>${hiveSwitchScriptPath}#${hiveSwitchScript}</file>
       <capture-output/>
       </shell>
       <ok to="master-decision"/>
       <error to="fail"/>
   </action>

回頁首

Oozie 電子郵件操作

Oozie 電子郵件操作（如清單 14 所示）可以從工作流應用程式傳送電子郵件。除了主題和訊息正文之外，電子郵件操作還必須擁有 to 和 cc（可選）地址。您可以使用逗號分割的電子郵件地址向多個收件人傳送電子郵件。電子郵件操作可同步執行，而工作流作業一直要等到傳送電子郵件之後才會觸發下一個操作。您可以使用 Hadoop EL 表示式將引數分配給電子郵件操作。

清單 14. Oozie 電子郵件操作

<action name="email">
  	<email xmlns="uri:oozie:email-action:0.1">
            <to>[email protected]</to>
            <subject>Oozie workflow finished successfully!</subject>
            <body>${wf:actionData('jProperties')['name']} | 
            ${wf:actionData('jProperties')['address']}</body>
        </email>
        <ok to="end"/>
        <error to="fail"/>
   </action>

回頁首

結束語

當多個相互依賴的作業與資料流捆綁在一起的時候，Oozie 工作流就會變成一個數據管道應用程式。Apache Oozie 工作流促進了資料邏輯流程的設計、錯誤處理、故障轉移機制等。您可以配置 Oozie 協調器或捆綁的應用程式來有效地管理工作流，不過，關於這些主題的討論已超出了本文的討論範圍。一些等效的 Hadoop 工作流引擎包括 Amazon Data Pipeline、Simple Workflow Engine、Azkaban、Cascading 和 Hamake。雖然 Hamake 和 Oozie 是基於 XML 的配置，但 Azkaban 是使用包含鍵值對的文字檔案進行配置的，而 Cascading 是使用 Java API 進行配置的。

http://www.ibm.com/developerworks/cn/data/library/bd-hadoopoozie/