BT

如何利用碎片时间提升技术认知与能力? 点击获取答案

扩展Oozie

| 作者 Boris Lublinsky 关注 1 他的粉丝 , Mike Segel 关注 0 他的粉丝 ,译者 侯伯薇 关注 0 他的粉丝 发布于 2011年10月21日. 估计阅读时间: 35 分钟 | AICon 关注机器学习、计算机视觉、NLP、自动驾驶等20+AI热点技术和最新落地成功案例。

在前面的两篇文章中[12],我们描述了Oozie工作流服务器,并且展示了几个工作流的示例。我们还描述了针对Oozie的工作流的部署和配置,以及用来启动、停止和监控Oozie工作流的工具。

在本文中,我们会向你展示Oozie的可扩展性,并说明它是如何支持我们实现自定义的、协同工作的语言扩展。

为什么需要自定义节点(Custom Node)?

正如我们在文章[1]中所说明的,Oozie之所以与众不同,是因为它提供了一种“最小化”的工作流语言,其中只包含少数几种控制和动作节点。尽管其中的一种动作节点是java动作节点,它让我们可以从Oozie工作流调用任意一个带有main方法的java类,但这种方法并非总是最佳的。原因之一就在于,java动作是在Hadoop簇集中作为map-reduce作业执行的,并且只带有唯一的Mapper任务。一方面,这带来了很多好处:

  • 它拥有内建的可伸缩性以及对map/reduce框架的灾难恢复支持,这样我们就不必这些特性构建到Oozie中。
  • 外部执行机制,这让Oozie引擎变得更轻量级,从而可以支持更多并发运行的过程。

另一方面,这种方法也有一些缺点:

  • 它把每个java节点都作为mapper任务启动,这会导致在Hadoop簇集中启动新的JVM而产生额外的开销。
  • 在外部执行java类导致了额外的网络传输,用于与Oozie服务器同步这些执行结果。
  • 从java节点传递参数成了非常耗费资源的操作。

尽管如此,在运行时间相对较长(几分钟甚至几小时)的map/reduce或者Pig作业中,好处会大大超过负载的缺点,但是,在简单的java节点中(参见[2]),我们就需要注意外部执行所导致的开销了。所以,使用自定义动作节点的原因之一,就是为了支持在Oozie的执行上下文中直接执行轻量级的java类[1]

使用自定义动作的另一个原因是为了提高工作流的语义和可读性。由于Oozie是一种支持基于Hadoop处理组件的工作流引擎,所以它的语法完全是以Hadoop执行为中心的——Hadoop文件系统、map/reduce、Pig等等。这种语法能够很好地符合Hadoop开发者的习惯,但是并没有涉及到太多关于给定动作的功能信息。我们可以为动作本身制定与业务相关的命名转换规则,但这只是特别用来解决问题的——对动作的命名只反映了给定过程的语法,既不是总体上的主题领域,也不能解决动作参数的问题,那些问题仍然只能由开发者来解决。

幸运的是,Oozie支持非常棒的扩展机制——自定义动作节点[3],它让我们可以很容易地解决这两个问题。自定义的动作节点让我们可以使用附加的动作(动词)来扩展Oozie的语言。Oozie的动作节点可以是同步的,也可以是异步的。

  • 同步节点——它在Oozie内部执行,在继续执行之前会等待这些节点完成动作。这些节点是为轻量级的任务所用的,像自定义计算,文件系统的移动、创建目录、删除等等。
  • 异步节点——它是由Oozie启动的,但是在Oozie引擎的外部执行,它会监控正在执行的动作,直到完成。这是通过动作的回调或者Oozie针对动作状态的polling操作完成的。

实现Oozie自定义动作处理程序

在这个例子中,我们会对独立的邮件程序进行转换,并展现到自定义的email动作中[2](代码1)。

package com.navteq.assetmgmt.oozie.custom;

import java.util.Properties;
import java.util.StringTokenizer;

import javax.mail.Message;
import javax.mail.Session;
import javax.mail.Transport;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;

import org.apache.oozie.ErrorCode;
import org.apache.oozie.action.ActionExecutor;
import org.apache.oozie.action.ActionExecutorException;
import org.apache.oozie.action.ActionExecutorException.ErrorType;
import org.apache.oozie.client.WorkflowAction;
import org.apache.oozie.util.XmlUtils;
import org.jdom.Element;
import org.jdom.Namespace;

public class EmailActionExecutor extends ActionExecutor {

         private static final String NODENAME = "eMail";

         private static final String SUCCEEDED = "OK";
  private static final String FAILED = "FAIL";
  private static final String KILLED = "KILLED";

         private static final String DEFAULMAILSERVER = "imailchi.navtech.com";
         private static final String EMAILSERVER = "emailServer";
         private static final String SUBJECT = "emailSubject";
         private static final String MESSAGE = "emailBody";
         private static final String FROM = "emailFrom";
         private static final String TO = "emailTo";

  public EmailActionExecutor() {
    super(NODENAME);
}

         @Override
         public void check(Context context, WorkflowAction action) throws ActionExecutorException {

                  // Should not be called for synch operation
                  throw new UnsupportedOperationException();
         }

         @Override
         public void end(Context context, WorkflowAction action)throws ActionExecutorException {

                  String externalStatus = action.getExternalStatus();
                  WorkflowAction.Status status = externalStatus.equals(SUCCEEDED) ?
                         WorkflowAction.Status.OK : WorkflowAction.Status.ERROR;
                  context.setEndData(status, getActionSignal(status));
         }

         @Override
         public boolean isCompleted(String arg0) {

                  return true;
         }

         @Override
         public void kill(Context context, WorkflowAction action) throws ActionExecutorException {

context.setExternalStatus(KILLED);
context.setExecutionData(KILLED, null);
         }

         @Override
         public void start(Context context, WorkflowAction action) throws ActionExecutorException {

                  // Get parameters from Node configuration
                  try{
                           Element actionXml = XmlUtils.parseXml(action.getConf());
                           Namespace ns = Namespace.getNamespace("uri:custom:email-action:0.1");

                           String server = actionXml.getChildTextTrim(EMAILSERVER, ns);
                           String subject = actionXml.getChildTextTrim(SUBJECT, ns);
                           String message = actionXml.getChildTextTrim(MESSAGE, ns);
                           String from = actionXml.getChildTextTrim(FROM, ns);
                           String to = actionXml.getChildTextTrim(TO, ns);

                           // Check if all parameters are there
                           if(server == null)
                                    server = DEFAULMAILSERVER;
                           if((message == null) || (from == null) || (to == null))
                                    throw new ActionExecutorException(ErrorType.FAILED,
   ErrorCode.E0000.toString(), "Not all parameters are defined");

                           // Execute action synchronously
                           SendMail(server, subject, message, from, to);
                           context.setExecutionData(SUCCEEDED, null);
                }
                catch(Exception e){
     context.setExecutionData(FAILED, null);
                       throw new ActionExecutorException(ErrorType.FAILED,
   ErrorCode.E0000.toString(), e.getMessage());
                }
     }

     // Sending an email
     public void SendMail(String server, String subject, String message,
                           String from, String to) throws Exception {

                // create some properties and get the default Session
                Properties props = new Properties();
                props.setProperty("mail.smtp.host", server);
                Session session = Session.getDefaultInstance(props, null);

                // create a message
                Message msg = new MimeMessage(session);

                // set the from and to address
                InternetAddress addressFrom = new InternetAddress(from);
                msg.setFrom(addressFrom);

                // To is a comma separated list
                StringTokenizer st = new StringTokenizer(to, ",");
                String [] recipients = new String[st.countTokens()];
                int rc = 0;
                while(st.hasMoreTokens())
                        recipients[rc++] = st.nextToken();
                InternetAddress[] addressTo = new InternetAddress[recipients.length];
                for (int i = 0; i < recipients.length; i++){
                        addressTo[i] = new InternetAddress(recipients[i]);
                }
                msg.setRecipients(Message.RecipientType.TO, addressTo);

                // Setting the Subject and Content Type
                msg.setSubject(subject);
                msg.setContent(message, "text/plain");
                Transport.send(msg);
     }
}

代码1: Email自定义动作

这个实现对ActionExecutor[2]类(由Oozie提供)进行了扩展,并重写了一些必要的方法。因为邮件的发送过程是一种非常快速的操作,所以我们决定将其实现为同步的动作处理程序,那意味着它会在Oozie的执行上下文中执行。

我们的实现(代码1)遵循了Oozie文档并实现了所有必需的方法:

  • 对于所有自定义动作处理程序都需要没有参数的构造函数。这个构造函数会注册动作处理程序的名称(使用动作名称来调用父类),我们会在工作流XML中使用它。
  • 我们可以使用InitActionType[3]方法来注册执行动作时可能发生的异常,以及它们的类型和错误信息,并为执行程序本身执行初始化操作。
  • Start方法是用来启动动作的执行操作的。因为我们实现的是同步动作,所以整个动作都会在此执行。这个方法是由Oozie调用的,它有两个参数:Context和WorkflowAction。Context参数提供了对Oozie工作流执行上下文的访问,其中主要包含了工作流的变量,并提供了对其进行操作的非常简单的API(set、get)[4]。而WorkflowAction则提供了Oozie对当前动作的定义。
  • 还有Check方法,Oozie会使用它来检查动作的状态。它不能作为同步动作来调用。
  • Kill方法,可以用来中断运行的作业或者动作。
  • End方法,可以用于所有清理动作,或者用于可能在完成动作之后所要做的处理。它还需要设置执行的结果。

部署并使用Oozie自定义动作处理程序

实现了自定义的动作执行方式之后,我们需要为新的email动作定义XML模式[5](代码2)。

<?xml version="1.0" encoding="UTF-8"?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema" 

                 xmlns:email="uri:custom:email-action:0.1"
     elementFormDefault="qualified"
     targetNamespace="uri:custom:email-action:0.1">
<xs:complexType name="EMAIL">

       <xs:sequence>
                 <xs:element name="emailServer" type="xs:string" minOccurs="0" maxOccurs="1" />

                 <xs:element name="emailSubject" type="xs:string" />
                 <xs:element name="emailFrom" type="xs:string" />

                 <xs:element name="emailTo" type="xs:string" />
                 <xs:element name="emailBody" type="xs:string" />

       </xs:sequence>
</xs:complexType>

<xs:element name="eMail" type="email:EMAIL"></xs:element>
</xs:schema>

代码2: 为email组件所用的XML schema

自定义动作节点和XML schema文件都需要打包在单独的jar文件中,比方说emailAction.Jar。我们可以使用Oozie的oozie-setup.sh脚本执行下面的命令,从而把这个(以及其他所有)jar文件添加到Oozie的war文件中。

$ bin/oozie-setup.sh -jars emailAction.jar:mail.jar (See Adding Jars to Oozie) 

代码3: 部署命令

向Oozie添加Jar文件

你要知道,Cloudera推荐的oozie-setup.sh命令行会重新构建你的war文件,并且, 如果你使用网页来监控作业,那么就会丢失java的脚本扩展。在测试方面,我们难以同时包含-extjs和-jars选项。作为权宜之计,我们会把jar文件复制到${CATALINA_BASE}/webapps/oozie/WEB-INF/lib中,其中${CATALINA_BASE}代表的是/var/lib/oozie/oozie-server。请注意,这里存在一定的风险,如果其他人重新构建了war文件,那么你就会丢失这些扩展,并且他们以后需要手动添加。对于测试来说,我们建议复制jar文件,然而,对于生产环境,我们建议把jar添加到war文件中。

现在我们需要把关于自定义执行器的信息注册到Oozie的运行时中。这是通过扩展oozie-site.xml完成的[6]。我们可以通过在Oozie配置文件oozie-site.xml中添加或者修改“oozie.service.ActionService.executor.ext.classes”[7]来注册自定义动作本身(代码4)。

……………………………………
<property>
       <name>oozie.service.ActionService.executor.ext.classes</name>
       <value>com.navteq.assetmgmt.oozie.custom. EmailActionExecutor </value>
</property>
……………………………………

代码4: 自定义执行配置

为新动作(代码2)所用的XML schema应该添加到oozie-site.xml中,位于属性“oozie.service.WorkflowSchemaService.ext.schemas”[8]之下(代码5)。

………………………………………
<property>
       <name>oozie.service.SchemaService.wf.ext.schemas</name>
       <value> emailAction.xsd</value>
</property>
…………………………………

代码5: 自定义模式配置

最后,Tomcat启动之后,我们就可以在工作流中使用自定义的动作节点了。

为了测试我们的实现,我们创建了简单的工作流(代码6),它会使用我们的执行器来发送email。

<!--
Copyright (c) 2011 NAVTEQ! Inc. All rights reserved.
Test email Oozie Script
-->
<workflow-app xmlns='uri:oozie:workflow:0.1' name='emailTester'>
 <start to='simpleEmail'/>

 <action name='simpleEmail'>
   <eMail xlmns=“ uri:custom:email-action:0.1”>
     <emailSubject>test</emailSubject>
     <emailFrom>mike.segel@<mycompany>.com</emailFrom>

     <emailTo>boris.lublinsky@<mycompany>.com</emailTo>
     <emailMessage>This is a test message, if you can see this, Mikey did something right! :)</emailMessage>
   </eMail>
  <error to="fail"/>

  <ok to="end"/>
 </action>
 <kill name="fail">
   <message>Workflow failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>

 </kill>
 <end name='end'/>
</workflow-app>

代码6: 使用email自定义动作执行器的简单工作流

结论

在这篇文章中,我们展示了如何通过创建自定义的动作执行器来扩展Oozie。这样做让我们可以定义和实现部门或者企业专用的Oozie语言(领域专用语言),其中具有部门或者企业的功能。这样的领域专用语言能够简化特定部门或者企业的构建过程,并提高代码的可读性。

尽管Oozie相对还不够成熟,但是它已经为处理包含多个map/reduce作业、并且能够向总体的业务过程添加非map-reduce作业的过程提供了基本的框架。随着越来越多的用户使用Oozie并提供反馈,我们相信它有足够的潜力,可以成为Hadoop环境中强大的集成部分。

对于Oozie来说,还有很多我们没有在这三篇文章中包含的内容。我们只是期望它们可以引起你对Oozie工作流引擎的足够兴趣,并且能够成为进一步研究Oozie的不错的起点。

参考信息

  1. Boris Lublinsky, Mike Segel. Introduction to Oozie.
  2. Boris Lublinsky, Mike Segel. Oozie by Example
  3. Oozie custom Action Nodes 
  4. Oozie source code 

关于作者

Boris Lublinsky是NAVTEQ公司的首席架构师,在这家公司中他的工作是为大型数据管理和处理、SOA以及实现各种NAVTEQ的项目定义架构的愿景。他还是InfoQ的SOA编辑,以及OASIS的SOA RA工作组的参与者。Boris是一位作者,还经常发表演讲,他最新的一本书是《Applied SOA》。

Michael Segel在过去二十多年间一直与客户写作,识别并解决他们的业务问题。Michael已经作为多种角色、在多个行业中工作过。他是一位独立顾问,总是期望能够解决所有有挑战的问题。Michael拥有俄亥俄州立大学的软件工程学位。


[1] 这种类的例子可能是各种计数器操作、简单的计算等等。

[2] 所有Oozie动作执行器都是Oozie分发程序的一部分,它们都是通过扩展这个类实现的。

[3] 在我们的实现中,我们使用的是默认的实现,这样就是为什么没有在代码中展现的原因。你可以查看一下Oozie的源代码[4],就可以知道在现存的Oozie动作处理程序中是如何实现这个方法的。

[4] 配置自定义执行器有两种方式——工作流变量和/或动作配置。在我们的例子中展示的是后者,但是在实际情况中,实际上总是二者的组合。

[5] 确保不仅要定义复杂的类型,还要对元素进行定义。那是Oozie所期望的。

[6] 通常在Oozie的分发包中叫做oozie-default.xml。

[7] 对于多个执行器,类名应该以逗号分隔。

[8] 对于多次执行,我们可以为多种模式使用逗号分隔的列表。

查看英文原文:Extending Oozie


给InfoQ中文站投稿或者参与内容翻译工作,请邮件至editors@cn.infoq.com。也欢迎大家加入到InfoQ中文站用户讨论组中与我们的编辑和其他读者朋友交流。

评价本文

专业度
风格

您好,朋友!

您需要 注册一个InfoQ账号 或者 才能进行评论。在您完成注册后还需要进行一些设置。

获得来自InfoQ的更多体验。

告诉我们您的想法

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我
社区评论

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

讨论

登陆InfoQ,与你最关心的话题互动。


找回密码....

Follow

关注你最喜爱的话题和作者

快速浏览网站内你所感兴趣话题的精选内容。

Like

内容自由定制

选择想要阅读的主题和喜爱的作者定制自己的新闻源。

Notifications

获取更新

设置通知机制以获取内容更新对您而言是否重要

BT