一个简单的使用Quartz和Oozie调度作业给大数据计算平台执行

简介:

一,介绍

Oozie是一个基于Hadoop的工作流调度器,它可以通过Oozie Client 以编程的形式提交不同类型的作业,如MapReduce作业和Spark作业给底层的计算平台(如 Cloudera Hadoop)执行。

Quartz是一个开源的调度软件,它为任务的调度执行提供了各种触发器以及监听器

下面使用Quartz + Oozie 将一个MapReduce程序提交给Cloudera Hadoop执行

二,调度思路

①为什么要用Quartz呢?主要是借助Quartz强大的触发器功能。它可以允许满足不同的调度需求,如每周执行作业一次、重复执行作业多少次。这里有一个重要的问题:假设我有一个作业需要重复执行,当第一次把该作业提交到CDH上执行后,以后需要执行该作业时不再是又一次把该作业上传到CDH上然后执行,而是把提交过的作业记录下来,下次需要运行时,直接让CDH再运行该作业。

②使用Quartz还有一个好处就是:在作业提交的时候可以做一些控制。比如,某种类型的作业提交的频率很高,或者运行时间较短(根据它上次执行完的情况来判断),那么下次运行它时,让它具有更高的优先级。

③使用Oozie的目的很明确,就是让它把作业发送给底层的计算平台,如CDH去执行作业。

三,Eclipse开发环境搭建

主要是需要Quartz和Oozie的依赖包。具体如下:

 四,实现思路

a) 调度系统目前只考虑调度两种类型的作业:Mapreduce作业和Spark作业。先把这二种作业通过Quartz传递给Oozie,然后再让Oozie把作业提交给CDH计算平台去执行。

b) Quartz提供了一个公共的Job接口。里面只有一个execute()方法,该方法负责完成Quartz所调度的作业的具体功能:把作业传递给Oozie

c) 定义一个抽象类BaseJob,它里面定义了二个方法。这二个方法主要是用来做一些准备工作,即使用Quartz把作业传递给Oozie时需要找到作业在HDFS上的存储目录,并将之复制执行目录下。

d) 最后是两个具体的实现类,MRJob和SparkJob,它们分别代表Mapreduce作业和Spark作业。在实现类里面完成作业的配置,然后将作业提交到CDH计算平台上执行。

相关类图如下:

 

五,具体代码分析

MRJob.java

实现了org.quartz.Job接口的execute(),该方法当触发器被触发时,会自动地被Quartz Schedule 调度执行。这样,就可以根据需要定义触发器,控制作业何时提交给Oozie。

复制代码
 1 @Override
 2     public void execute(JobExecutionContext arg0) throws JobExecutionException {
 3         try{
 4             String jobId = wc.run(conf);
 5             System.out.println("Workflow job submitted");//submit job to oozie and get the jobId
 6             
 7             //wait until the workflow job finishes
 8             while(wc.getJobInfo(jobId).getStatus() == Status.RUNNING){
 9                 System.out.println("Workflow job running...");
10                 try{
11                     Thread.sleep(10*1000);
12                 }catch(InterruptedException e){e.printStackTrace();}
13             }
14             System.out.println("Workflow job completed!");
15             System.out.println(wc.getJobId(jobId));
16         }catch(OozieClientException e){e.printStackTrace();}
17     }
复制代码

 

测试的main函数程序如下:可以看出对于客户端而言,只需要按照编写常规的Quartz作业方式,就可以调试MapReduce作业了。要想运行该程序,当然还得提前准备到作业的运行环境。具体参考

复制代码
 1 import static org.quartz.JobBuilder.newJob;
 2 import static org.quartz.TriggerBuilder.newTrigger;
 3 
 4 import java.util.Date;
 5 
 6 import org.quartz.JobDetail;
 7 import org.quartz.Scheduler;
 8 import org.quartz.SchedulerFactory;
 9 import org.quartz.SimpleTrigger;
10 import org.quartz.impl.StdSchedulerFactory;
11 import org.slf4j.Logger;
12 import org.slf4j.LoggerFactory;
13 
14 import com.quartz.job.MRJob;
15 
16 
17 public class QuartzOozieJobTest {
18     public static void main(String[] args) throws Exception{
19         QuartzOozieJobTest test = new QuartzOozieJobTest();
20         test.run();
21     }
22     
23     public void run() throws Exception{
24         Logger log = LoggerFactory.getLogger(QuartzOozieJobTest.class);
25 
26         log.info("------- Initializing ----------------------");
27 
28         SchedulerFactory sf = new StdSchedulerFactory();
29         Scheduler sched = sf.getScheduler();
30         
31         long startTime = System.currentTimeMillis() + 20000L;
32         Date startTriggerTime = new Date(startTime);
33         
34         JobDetail jobDetail = newJob(MRJob.class).withIdentity("job", "group1").build();
35         SimpleTrigger trigger = (SimpleTrigger) newTrigger().withIdentity("trigger", "group1").startAt(startTriggerTime).build();
36         
37         Date ft = sched.scheduleJob(jobDetail, trigger);
38         
39         log.info(jobDetail.getKey() + " will submit at " + ft + " only once.");
40         
41         sched.start();
42 //        sched.shutdown(true);
43     }
44 }
复制代码

 

整个项目的源代码下载


本文转自hapjin博客园博客,原文链接:http://www.cnblogs.com/hapjin/p/4943592.html,如需转载请自行联系原作者

相关实践学习
基于MaxCompute的热门话题分析
Apsara Clouder大数据专项技能认证配套课程:基于MaxCompute的热门话题分析
相关文章
|
11月前
|
数据可视化 关系型数据库 MySQL
基于python大数据的的海洋气象数据可视化平台
针对海洋气象数据量大、维度多的挑战,设计基于ECharts的可视化平台,结合Python、Django与MySQL,实现数据高效展示与交互分析,提升科研与决策效率。
|
数据采集 人工智能 大数据
10倍处理效率提升!阿里云大数据AI平台发布智能驾驶数据预处理解决方案
阿里云大数据AI平台推出智能驾驶数据预处理解决方案,助力车企构建高效稳定的数据处理流程。相比自建方案,数据包处理效率提升10倍以上,推理任务提速超1倍,产能翻番,显著提高自动驾驶模型产出效率。该方案已服务80%以上中国车企,支持多模态数据处理与百万级任务调度,全面赋能智驾技术落地。
1653 0
|
11月前
|
传感器 人工智能 监控
拔俗多模态跨尺度大数据AI分析平台:让复杂数据“开口说话”的智能引擎
在数字化时代,多模态跨尺度大数据AI分析平台应运而生,打破数据孤岛,融合图像、文本、视频等多源信息,贯通微观与宏观尺度,实现智能诊断、预测与决策,广泛应用于医疗、制造、金融等领域,推动AI从“看懂”到“会思考”的跃迁。
779 0
|
SQL 存储 分布式计算
ODPS技术架构深度剖析与实战指南——从零开始掌握阿里巴巴大数据处理平台的核心要义与应用技巧
【10月更文挑战第9天】ODPS是阿里巴巴推出的大数据处理平台,支持海量数据的存储与计算,适用于数据仓库、数据挖掘等场景。其核心组件涵盖数据存储、计算引擎、任务调度、资源管理和用户界面,确保数据处理的稳定、安全与高效。通过创建项目、上传数据、编写SQL或MapReduce程序,用户可轻松完成复杂的数据处理任务。示例展示了如何使用ODPS SQL查询每个用户的最早登录时间。
2147 1
|
消息中间件 分布式计算 大数据
大数据-123 - Flink 并行度 相关概念 全局、作业、算子、Slot并行度 Flink并行度设置与测试
大数据-123 - Flink 并行度 相关概念 全局、作业、算子、Slot并行度 Flink并行度设置与测试
782 0
|
分布式计算 算法 大数据
大数据时代的智能研发平台需求与阿里云DIDE的定位
阿里云DIDE是一站式智能大数据开发与治理平台,致力于解决传统大数据开发中的效率低、协同难等问题。通过全面整合资源、高度抽象化设计及流程自动化,DIDE显著提升数据处理效率,降低使用门槛,适用于多行业、多场景的数据开发需求,助力企业实现数字化转型与智能化升级。
685 1
|
机器学习/深度学习 存储 算法
Java 大视界 -- Java 大数据在智能农业无人机植保作业路径规划与药效评估中的应用(165)
本文围绕 Java 大数据在智能农业无人机植保作业路径规划与药效评估中的应用展开,剖析作业现状与挑战,阐述技术原理及应用方法,结合案例与代码,给出具有实操性的解决方案。
Java 大视界 -- Java 大数据在智能农业无人机植保作业路径规划与药效评估中的应用(165)
|
存储 SQL 大数据
【重磅发布】AllData数据中台核心功能:湖仓一体化平台
杭州奥零数据科技有限公司成立于2023年,专注于数据中台业务,维护开源项目AllData并提供商业版解决方案。AllData提供数据集成、存储、开发、治理及BI展示等一站式服务,支持AI大模型应用,助力企业高效利用数据价值。
【重磅发布】AllData数据中台核心功能:湖仓一体化平台
|
SQL 存储 HIVE
鹰角基于 Flink + Paimon + Trino 构建湖仓一体化平台实践项目
本文整理自鹰角网络大数据开发工程师朱正军在Flink Forward Asia 2024上的分享,主要涵盖四个方面:鹰角数据平台架构、数据湖选型、湖仓一体建设及未来展望。文章详细介绍了鹰角如何构建基于Paimon的数据湖,解决了Hudi入湖的痛点,并通过Trino引擎和Ranger权限管理实现高效的数据查询与管控。此外,还探讨了湖仓一体平台的落地效果及未来技术发展方向,包括Trino与Paimon的集成增强、StarRocks的应用以及Paimon全面替换Hive的计划。
2134 1
鹰角基于 Flink + Paimon + Trino 构建湖仓一体化平台实践项目
|
SQL 存储 HIVE
鹰角基于 Flink + Paimon + Trino 构建湖仓一体化平台实践项目
鹰角基于 Flink + Paimon + Trino 构建湖仓一体化平台实践项目
1124 2