1.1 Mapreduce作业流JobControl和Oozie
更复杂的任务,需要多个mapreduce作业,形成作业流,而不是增加map和reduce的复杂度。复杂问题,可以用高级语言pig、hive、cascading、crunch、spark。
1.1.1 问题分解成mapreduce作业流
按天统计每天的最高气温,求出每月的最高气温平均值;再找出每年中平均气温最高的月,找出1901-2000年哪一年的哪一个月的平均气温最高。分解为mapreduce作业流。
Mapreduce作业1
(1) map计算出每天的最高气温;(日期,最高气温)
(2) reduce计算所有每月最高气温的平均值。
MapReduce作业2
(1) 找出每年平均气温最高的月。(年份月份,最高平均气温)
(2) 找出1901-2000年之间气温最高的年月。
1.1.2 JobControl控制作业流顺序
可以使用jobClient线性执行两个作业,并通过waitForcompletion()返回值判断执行是否成功。
JobClient.runJob(conf1);
JobClient.runJob(conf2);
还可以用org.apache.hadoop.mapreduce.jobcontrol包或者org.apache.hadoop.mapred.jobcontrol中的JobControl实例来控制执行顺序。可以查看进程和作业状态,错误信息。使用步骤
(1)作业1加入控制器
ControlledJob ctrljob1 = new ControlledJob(conf);
ctrljob1.setJob(job1);
(2)作业2加入控制器,并依赖于作业1
ControlledJob ctrljob2 = new ControlledJob(conf);
ctrljob2.setJob(job2);
// 设置多个作业直接的依赖关系,意思为job2的启动,依赖于job1作业的完成
ctrljob2.addDependingJob(ctrljob1);
(3) 作业3加入控制容器,并能个依赖于作业2
ControlledJob ctrljob3 = new ControlledJob(conf);
ctrljob3.setJob(job3);
ctrljob3.addDependingJob(ctrljob2);
(4)JobControl控制作业流
// 主的控制容器,控制上面的总的3个子作业
JobControl jobCtrl = new JobControl("myctrl");
// 添加到总的JobControl里,进行控制
jobCtrl.addJob(ctrljob1);
jobCtrl.addJob(ctrljob2);
jobCtrl.addJob(ctrljob3);
// 在线程启动,记住一定要有这个
Thread t = new Thread(jobCtrl);
t.start();
while (true)
{
if (jobCtrl.allFinished())
{// 如果作业成功完成,就打印成功作业的信息
System.out.println(jobCtrl.getSuccessfulJobList());
System.out.println("所有job执行完毕");
jobCtrl.stop();
break;
}
}
1.1.3 Apache Oozie工作流调度系统
http://oozie.apache.org/docs/5.0.0/index.html
Oozie是工作流调度系统,用于调度作业流按照复杂的顺序逻辑执行。先将作业组成工作流workflow,然后coordinate引擎用于协调多个工作流workflow执行。Bundle将多个coordinate进行汇总处理。Workflow->Coordinate->Bundle.
Oozie与支持多种类型的Hadoop作业(如Java map-reduce、流式map-reduce、Pig、Hive、Sqoop和Distcp)以及特定于系统的工作(如Java程序和shell脚本)。
作业流由actions 集合(例如Hadoop map/reduce作业,pig作业),actions被安排在一个控制依赖项DAG(Direct Acyclic Graph)中,按顺序执行。
workflow工作流 |
流式作业,将任务分解为hadoop作业(MapReduce,pig,Hive)。通过workflow.xml配置执行工作流。 |
coordinator引擎 |
周期性执行工作流或者控制相互依赖的workflow 作业流。 |
Bundle |
将多个coordinate进行汇总处理。 |
Oozie架构图
Workflow流程图
Coordinate流程图
Bundle流程图
1.1.4 oozie工作流job组成
一个oozie 的 job 一般由以下文件组成:
job.properties :记录了job的属性,nameNode,jobTracker和 workflow.xml在hdfs中的位置必须设置。
workflow.xml :使用xml配置文件定义任务的流程和分支
lib目录:用来执行具体的任务。
(1)job.properties属性
KEY |
含义 |
nameNode |
HDFS地址 |
jobTracker |
jobTracker(ResourceManager)地址 |
queueName |
Oozie队列(默认填写default) |
examplesRoot |
全局目录(默认填写examples) |
oozie.usr.system.libpath |
是否加载用户lib目录(true/false) |
oozie.libpath |
用户lib库所在的位置 |
oozie.wf.application.path |
Oozie流程所在hdfs地址(workflow.xml所在的地址) |
user.name |
当前用户 |
oozie.coord.application.path |
Coordinator.xml地址(没有可以不写) |
oozie.bundle.application.path |
Bundle.xml地址(没有可以不写) |
实例如下
nameNode=hdfs://cm1:8020
jobTracker=cm1:8032
queueName=default
examplesRoot=examples
oozie.wf.application.path=${nameNode}/user/workflow/oozie/shell
(2)workflow.xml
Workflow.xml定义了控制节点(start,kill,end)和动作节点action,控制节点控制流程的运行,动作节点定义了作业的任务。
Wordcount的workflow.xml实例
<workflow-app name='wordcount-wf' xmlns="uri:oozie:workflow:0.1">
<start to='wordcount'/>
<action name='wordcount'>
<map-reduce>
<job-tracker>${jobTracker}</job-tracker>
<name-node>${nameNode}</name-node>
<configuration>
<property>
<name>mapred.mapper.class</name>
<value>org.myorg.WordCount.Map</value>
</property>
<property>
<name>mapred.reducer.class</name>
<value>org.myorg.WordCount.Reduce</value>
</property>
<property>
<name>mapred.input.dir</name>
<value>${inputDir}</value>
</property>
<property>
<name>mapred.output.dir</name>
<value>${outputDir}</value>
</property>
</configuration>
</map-reduce>
<ok to='end'/>
<error to='end'/>
</action>
<kill name='kill'>
<message>Something went wrong: ${wf:errorCode('wordcount')}</message>
</kill/>
<end name='end'/>
</workflow-app>
(3)Lib目录:
在workflow工作流定义的同级目录下,需要有一个lib目录,在lib目录中存在java节点MapReduce使用的jar包。
需要注意的是,oozie并不是使用指定jar包的名称来启动任务的,而是通过指定主类来启动任务的。在lib包中绝对不能存在某个jar包的不同版本,不能够出现多个相同主类。
(4)执行作业流
将job.properties、workflow.xml和lib文件夹放在用户名的目录(${namenode}/user/${user.name})下的max-temp-workflow,复制到HDFS.
%hadoop fs –put Hadoop-examples/target/max-temp-workflow max-temp-workflow.
使用oozie命令行工具设置环境变量OOZIE_URL指定Oozie服务器
%export OOZIE_URL=”http://localhsot:11000/oozie”
运行工作流
% oozie job –config ch06-mr-dev/max-temp-workflow.preoperties –run job:0001-121
–config设置工作流属性文件,内含namenode、资源管理器地址、作业路径(oozie.wf.application.path=${namenode}/user/${user.name}/max-temp-workflow)
-run 参数用于运行指定的作业。
通过-info查看作业状态
%oozie job –info 0001-121