技术开发 频道

Java Web应用程序: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。

       注:[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管理页面。

0