自定义实现wordcount的workflow
- 在yarn上测试wordcount程序
$ bin/yarn jar /opt/datas/wd.jar mapreduce.WordCount /input /output
15.png
- 生产一个应用目录，并把wordcount的jar包放到应用目录中的lib目录下
$ cd /opt/modules/cdh/oozie-4.0.0-cdh5.3.6/
$ mkdir my-apps
$ cp -r examples/apps/map-reduce/ my-apps/
$ cp /opt/datas/wd.jar my-apps/map-reduce/lib/
- 修改job.properties文件
# 定义变量
nameNode=hdfs://bigdata-00:8020
jobTracker=bigdata-00:8032
queueName=default
oozieAppsRoot=user/wulei/my-apps
oozieDataRoot=user/wulei/oozie/datas
inputDir=map-reduce/input
outputDir=map-reduce/output
# 定义workflow.xml的路径
oozie.wf.application.path=${nameNode}/${oozieAppsRoot}/map-reduce/workflow.xml
- 修改workflow.xml文件
<workflow-app xmlns="uri:oozie:workflow:0.5" name="map-reduce-wf">
<start to="mr-node"/>
<action name="mr-node">
<map-reduce>
<job-tracker>${jobTracker}</job-tracker>
<name-node>${nameNode}</name-node>
<!-- 准备动作，如果输出路径存在，先进行删除-->
<prepare>
<delete path="${nameNode}/${oozieDataRoot}/${outputDir}"/>
</prepare>
<configuration>
<property>
<name>mapred.job.queue.name</name>
<value>${queueName}</value>
</property>
<!-- set map class and reduce class-->
<property>
<name>mapreduce.job.map.class</name>
<value>mapreduce.WordCount$WordCountMapper</value>
</property>
<property>
<name>mapreduce.job.reduce.class</name>
<value>mapreduce.WordCount$WordCountReducer</value>
</property>
<!-- set inputdir and outputdir-->
<property>
<name>mapreduce.input.fileinputformat.inputdir</name>
<value>${nameNode}/${oozieDataRoot}/${inputDir}</value>
</property>
<property>
<name>mapreduce.output.fileoutputformat.outputdir</name>
<value>${nameNode}/${oozieDataRoot}/${outputDir}</value>
</property>
<!-- 因为oozie会默认使用mapreduce1的api，这里需要设置为mapreduce新的api-->
<property>
<name>mapred.mapper.new-api</name>
<value>true</value>
</property>
<property>
<name>mapred.reducer.new-api</name>
<value>true</value>
</property>
<!-- set map output key and value class-->
<property>
<name>mapreduce.map.output.key.class</name>
<value>org.apache.hadoop.io.Text</value>
</property>
<property>
<name>mapreduce.map.output.value.class</name>
<value>org.apache.hadoop.io.LongWritable</value>
</property>
<!-- set reduceoutput key and value class-->
<property>
<name>mapreduce.job.output.key.class</name>
<value>org.apache.hadoop.io.Text</value>
</property>
<property>
<name>mapreduce.job.output.value.class</name>
<value>org.apache.hadoop.io.IntWritable</value>
</property>
</configuration>
</map-reduce>
<ok to="end"/>
<error to="fail"/>
</action>
<kill name="fail">
<message>Map/Reduce failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
</kill>
<end name="end"/>
</workflow-app>
- 在hdfs上创建目录，并上传input数据
$ bin/hdfs dfs -mkdir -p /user/wulei/oozie/datas/map-reduce/input
$ bin/hdfs dfs -put -put /opt/datas/a.txt oozie/datas/map-reduce/input
- 上传oozie-apps目录
$ bin/hdfs dfs -put /opt/modules/cdh/oozie-4.0.0-cdh5.3.6/my-apps/ my-apps
- 运行测试
$ bin/oozie job -oozie http://bigdata-00:11000/oozie -config my-apps/map-reduce/job.properties -run
-
查看测试结果
yarn任务执行界面
16.png
oozie的web界面
17.png
输出文件目录
18.png
自定义实现WorkFlow中shell action
- 拷贝默认的shell目录来进行修改
$ cp -r ./examples/apps/shell/ my-apps/
- 定义job.properties
nameNode=hdfs://bigdata-00:8020
jobTracker=bigdata-00:8032
queueName=default
examplesRoot=user/wulei/my-apps/shell
#定义workflow.xml路劲
oozie.wf.application.path=${nameNode}/${examplesRoot}/workflow.xml
#定义一个变量名存储shell文件名
EXEC=oozie.sh
- 定义workflow.properties
<workflow-app xmlns="uri:oozie:workflow:0.4" name="shell-wf">
<start to="shell-node"/>
<action name="shell-node">
<shell xmlns="uri:oozie:shell-action:0.2">
<job-tracker>${jobTracker}</job-tracker>
<name-node>${nameNode}</name-node>
<configuration>
<property>
<name>mapred.job.queue.name</name>
<value>${queueName}</value>
</property>
</configuration>
<exec>${EXEC}</exec>
<!-- 定义shell脚本路劲 -->
<file>${nameNode}/${examplesRoot}/${EXEC}#${EXEC}</file> <!--Copy the executable to compute node's current working directory -->
</shell>
<ok to="end"/>
<error to="fail"/>
</action>
<kill name="fail">
<message>Shell action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
</kill>
<end name="end"/>
</workflow-app>
-
定义要实现的脚本
19.png
-
上传到hdfs后运行
20.png
-
查看运行结果
21.png
22.png
Oozie coordinator调度
当有一个复杂的工作流job，希望每天定时执行，使用crontab方式调用需要编写大量的脚本，还要通过大量的判断来控制每个工作流job的执行顺序问题。Oozie中的Coordinator可以让每个工作流workflow当做一个动作来运行，这样就可以将多个job组织起来，并制定触发时间和频率。它的调度方式有两种：一种是基于时间调度；还有一种是基于数据可用性调度(判断数据是否存在和是否符合标准)。
-
如果要设置基于时间的调度，需要先配置时区，因为oozie默认使用的是UTC时区，服务器默认使用的是CST，我们把他们修改为GMT +0800（当然也可以是其他时区）
查看系统当前时区：
23.png
如果输出的时区不是+0800,需要进行修改
$ rm -rf /etc/localtime
$ ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
修改oozie时区，在oozie-sitexml添加参数
<property>
<name>oozie.processing.timezone</name>
<value>GMT+0800</value>
</property>
修改oozie/-server/webapps/oozie/oozie-console.js中的时区设置，不然会影响web界面的时间，修改完成后重启生效
function getTimeZone() {
Ext.state.Manager.setProvider(new Ext.state.CookieProvider());
return Ext.state.Manager.get("TimezoneId","GMT+0800");
}
- 拷贝模板中的coordinator案例
$ cp -r examples/apps/cron-schedule/ my-apps/ - 修改job.properties
nameNode=hdfs://bigdata-00:8020
jobTracker=bigdata-00:8032
queueName=default
examplesRoot=user/wulei/my-apps/cron-schedule
# 定义coordinate路径
oozie.coord.application.path=${nameNode}/${examplesRoot}
start=2016-11-13T12:30+0800
end=2016-11-13T13:30+0800
# 定义workflow路劲
workflowAppUri=${nameNode}/${examplesRoot}
- 修改coorndinate.xml，让这个任务从开始时间到结束时间内每分钟执行一次
<coordinator-app name="cron-coord" frequency="${coord:minutes(1)}" start="${start}" end="${end}" timezone="GMT+0800"
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>
</configuration>
</workflow>
</action>
</coordinator-app>
-
oozie的默认配置中，至少为每5分钟执行一次，所以需要修改默认配置
24.png
-
上传至hdfs后执行
25.png
-
查看执行结果
27.png
26.png
网友评论