oozie coordinator 定时调度

2604 0 0

作者:1573=

(本段内容摘自http://blog.sina.com.cn/s/blog_e699b42b0102xjqw.html Oozie总结行成于思的博客) Oozie提出了Coordinator的概念，它能够将每个工作流Job作为一个动作（Action）来运行，相当于工作流定义中的一个执行节点（我们可以理解为工作流的工作流），这样就能够将多个工作流Job组织起来，称为Coordinator Job，并指定触发时间和频率，还可以配置数据集、并发数等。一个Coordinator Job包含了在Job外部设置执行周期和频率的语义，类似于在工作流外部增加了一个协调器来管理这些工作流的工作流Job的运行。相比较于单个任务的配置，Coordinator的配置是在job.properties中将oozie.wf.application.path修改为oozie.coord.application.path。并且还要定义一个coordinator.xml

coordinator application:

　　 coordinator application是在满足一组条件时触发动作（通常是工作流作业）的程序。条件可以是时间频率、新数据集实例或其他外部事件。

　　 coordinator application的类型：同步：它的协调器动作是在指定的时间间隔创建的,通常是参数化的。

coordinator job:

　　要创建一个coordinator job,必须向协调器引擎提供解决所有coordinator application参数的作业配置.

　　coordinator job是从开始时间到结束时间运行的coordinator application的运行实例,开始时间必须比结束时间早。

　　通常情况下,一个coordinator 作业是下列状态之一:PREP, RUNNING, RUNNINGWITHERROR, PREPSUSPENDED, SUSPENDED, SUSPENDEDWITHERROR, PREPPAUSED, PAUSED, PAUSEDWITHERROR, SUCCEEDED, DONEWITHERROR, KILLED, FAILED .

　　有效的coordinator job状态转换是:

　当一个coordinator 提交作业，Oozie解析 coordinator job XML。Oozie然后创建与状态准备coordinator 记录并返回一个唯一的ID，如果没有设置暂停时间coordinator 也立即开始.

Coordinator Action:

　 coordinator job 创建并执行 coordinator actions.

　 coordinator action通常是一个workflow job ，它消耗并生成数据集实例。

　一旦创建了coordinator action（这也被称为正在实现的action）， coordinator action 将一直等待，直到满足执行所需的所有输入，或者直到等待超时为止。

Synchronous Coordinator Application定义:

　　 synchronous coordinator 是由 name, start time ,end time, the frequency of creation of its coordinator actions, the input events, the output events , action control information来定义的.

语法:　　

<coordinator-app name="[NAME]" frequency="[FREQUENCY]"
                    start="[DATETIME]" end="[DATETIME]" timezone="[TIMEZONE]"
                    xmlns="uri:oozie:coordinator:0.1">  　
#frequency:执行频率,小于五分钟要修改配置 start,end:开始与结束时间,若想跟北京时间一样也要修改配置文件,并修改时间格式

      <controls>
        <timeout>[TIME_PERIOD]</timeout>
        <concurrency>[CONCURRENCY]</concurrency>
        <execution>[EXECUTION_STRATEGY]</execution>
      </controls>
.
      <datasets>　　　　
        <include>[SHARED_DATASETS]</include>
        ...
.
        <!-- Synchronous datasets --> #---数据生成目录
        <dataset name="[NAME]" frequency="[FREQUENCY]"
                 initial-instance="[DATETIME]" timezone="[TIMEZONE]">
          <uri-template>[URI_TEMPLATE]</uri-template>
        </dataset>
        ...
.
      </datasets>
.
      <input-events>　　　　#----定义了数据触发条件
        <data-in name="[NAME]" dataset="[DATASET]">
          <instance>[INSTANCE]</instance>
          ...
        </data-in>
        ...
        <data-in name="[NAME]" dataset="[DATASET]">
          <start-instance>[INSTANCE]</start-instance>
          <end-instance>[INSTANCE]</end-instance>
        </data-in>
        ...
      </input-events>
      <output-events>
         <data-out name="[NAME]" dataset="[DATASET]">
           <instance>[INSTANCE]</instance>
         </data-out>
         ...
      </output-events>
      <action>
        <workflow>
          <app-path>[WF-APPLICATION-PATH]</app-path>　　　　#---workflow.xml所在hdfs目录
          <configuration>
            <property>    #----定义传给workflow的参数
              <name>[PROPERTY-NAME]</name>
              <value>[PROPERTY-VALUE]</value>
            </property>
            ...
         </configuration>
       </workflow>
      </action>
   </coordinator-app>

官网给出的例子:

<coordinator-app name="hello-coord" frequency="${coord:days(1)}"
                    start="2009-01-02T08:00Z" end="2009-01-02T08:00Z"
                    timezone="America/Los_Angeles"
                    xmlns="uri:oozie:coordinator:0.1">
      <datasets>
        <dataset name="logs" frequency="${coord:days(1)}"
                 initial-instance="2009-01-02T08:00Z" timezone="America/Los_Angeles">
          <uri-template>hdfs://bar:8020/app/logs/${YEAR}${MONTH}/${DAY}/data</uri-template>
        </dataset>
        <dataset name="siteAccessStats" frequency="${coord:days(1)}"
                 initial-instance="2009-01-02T08:00Z" timezone="America/Los_Angeles">
          <uri-template>hdfs://bar:8020/app/stats/${YEAR}/${MONTH}/${DAY}/data</uri-template>
        </dataset>
      </datasets>
      <input-events>　　　　
        <data-in name="input" dataset="logs">
          <instance>2009-01-02T08:00Z</instance>
        </data-in>
      </input-events>
      <output-events>
         <data-out name="output" dataset="siteAccessStats">
           <instance>2009-01-02T08:00Z</instance>
         </data-out>
      </output-events>
      <action>
        <workflow>
          <app-path>hdfs://bar:8020/usr/joe/logsprocessor-wf</app-path>    
          <configuration>
            <property>　　　
              <name>wfInput</name>
              <value>${coord:dataIn('input')}</value>
            </property>
            <property>
              <name>wfOutput</name>
              <value>${coord:dataOut('output')}</value>
            </property>
         </configuration>
       </workflow>
      </action>
   </coordinator-app>

我们工作时写的:

coordinator.xml

<coordinator-app name="cron-coord" frequency="${coord:minutes(6)}" start="${start}" end="${end}" timezone="Asia/Shanghai" 
xmlns="uri:oozie:coordinator:0.2">
    <action>
        <workflow>
            <app-path>${workflowAppUri}</app-path>
            <configuration>
                <property>
                    <name>jobTracker</name>
                    <value>${jobTracker}</value>
                </property>
                <property>
                    <name>nameNode</name>
                    <value>${nameNode}</value>
                </property>
                <property>
                    <name>queueName</name>
                    <value>${queueName}</value>
                </property>
                <property>
                    <name>mainClass</name>
                    <value>com.ocn.itv.rinse.ErrorCollectRinse</value>
                </property>
            </configuration>
        </workflow>
    </action>
</coordinator-app>

workflow.xml


<workflow-app  name="Spark-example1" xmlns="uri:oozie:workflow:0.5">  
    <start to="spark-SparkOozieAction"/>  
    <action name="spark-SparkOozieAction">  
       <spark xmlns="uri:oozie:spark-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>            
            <master>yarn-cluster</master>
            <mode>cluster</mode>
            <name>Spark Example</name>
            <class>${mainClass}</class>            
            <jar>ocn-itv-spark-3.0.3-rc1.jar</jar> 
            <spark-opts>${sparkopts}</spark-opts> 
            <arg>${input}</arg>            
        </spark >  
        <ok to="end"/>
        <error to="fail"/>    
    </action>  
  <kill name="fail">  
       <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>  
    </kill>  
   <end name="end"/>  
</workflow-app>

job.properties

nameNode=hdfs://hgdp-001:8020 
jobTracker=hgdp-001:8032
queueName=default
input=2017-05-09　　#输入参数
start=2017-08-28T17:50+0800  #开始时间
end=2017-08-28T18:50+0800　　#结束时间
sparkopts=--executor-memory 1G  
oozie.use.system.libpath=True　　#是否开启系统用户lib库(有action依赖的jar包)
hdfspath=user/root
examplesRoot=ocn-itv-oozie  #全局目录
oozie.libpath=${nameNode}/${hdfspath}/${examplesRoot}/lib/  #用户lib库地址(自添加jar包)
workflowAppUri=${nameNode}/${hdfspath}/${examplesRoot}/wf/wf1/  #workflow.xml所在目录
oozie.coord.application.path=${nameNode}/${hdfspath}/${examplesRoot}/cd/cd1/　　#coordinator.xml坐在目录

最后运行:

　　启动任务:oozie job -config job.properties -run -oozie http://xxxx(地址):11000/oozie

运行结果:

内容来源于网络如有侵权请私信删除

标签： R语言 Delphi erlang verilog dart

你还没有登录，请先登录或注册！

还没有人评论，欢迎说说您的想法！

oozie coordinator 定时调度