使用azkaban 建立一个简单的任务调度系统

Bmob 发布于2019-08-15 12:27 / 724人阅读

摘要：使用建立一个简单的任务调度系统可以作为任务的调度工具，也可以作为任务和任务的调度工具。更加方便，只要在中增加一个方法，返回对象，然后任务可以直接在初始化的中读取到这个参数。

使用azkaban 建立一个简单的任务调度系统

azkaban 可以作为hadoop 任务的调度工具，也可以作为shell任务和java任务的调度工具。安装过程有点繁琐，见安装文档。

Job Type

见链接 http://www.itwendao.com/artic...

任务DAG

通过任务之间的依赖关系(dependencies)构建DAG

下面是一个 upload.job 的例子, 依赖 report-en.job

type=java
#指定类的全路径
job.class=com.example.demo.task.BbUploadJob
#指定执行jar包的路径
classpath=lib/*
#依赖任务
dependencies=report-en
#jvm 参数
Xmx=512M
# 自定义参数
batch.timestamp=${azkaban.flow.start.timestamp}

schedule

azkaban 目前包含Quartz，支持 Cron 表达式

java job

JavaJob 目前需要配置Hadoop，但是可以通过修改源代码来去掉 Hadoop 的依赖关系，参考安装过程。

Java Job template

run 相当于 main 方法，任务代码写在这里

cancel 在 run 方法出现 Exception 之后调用，任务失败后处理

getJobGeneratedProperties 是输出的参数，用于给下一个任务传递参数

import azkaban.utils.Props;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class JavaJob1 {

    private static final Logger logger = LoggerFactory.getLogger(JavaJob1.class);

    private Props props;

    public JavaJob1(String name, Props props) {
        this.props = props;
    }
    
    public void run() {
        String timestamp = props.getString("azkaban.flow.start.timestamp");
        logger.info("timestamp value is ==> " + timestamp);
    }
    
    public void cancel(){
        
    }
    

    public Props getJobGeneratedProperties(){
        Props props = new Props();
        return props;
    }
}

java job 中使用 Spring

在JavaJob 中可以使用Spring，和普通的Java代码一模一样

this.classPathXmlApplicationContext = new ClassPathXmlApplicationContext("applicationContext.xml");
this.reportService = (ReportService) this.classPathXmlApplicationContext.getBean("reportService");

其他问题 邮件通知

azkaban 原生就支持通知功能，在安装的时候配置smtp服务器，在job文件中里配置failure.emails, success.emails, notify.emails 来通知任务执行情况(多个邮箱地址用逗号分隔)

任务之间参数传递

azkaban支持任务之间传递参数，A任务可以向依赖A的任务B传递参数。实际上是通过读写临时文件来实现这个功能。

System.getenv("JOB_OUTPUT_PROP_FILE") // 任务输出的参数文件
System.getenv("JOB_PROP_FILE")   // 任务初始化的参数文件

B任务初始化的参数文件中会包含A任务输出的参数，这是azkaban帮我们做的。

JavaJob 更加方便，只要在Job中增加一个getJobGeneratedProperties方法，返回Props对象，然后B任务可以直接在初始化的Props中读取到这个参数。

public Props getJobGeneratedProperties(){
    Props props = new Props();
    props.put("demo.test.arg1", "Hello World!");
    return props;
}

云服务器 GPU云服务器任务调度任务调度过程任务调度算法任务调度开源

文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。

转载请注明本文地址：https://www.ucloud.cn/yun/67963.html

cdhazkaban-cdhazkaban介绍、安装和使用

工作流调度系统一个完整的数据分析系统大都是由海量的任务单元组成shell脚本程序，java，mapreduce程序、hive脚本等等组成，非常复杂，各任务单元之间存在时间先后及前后依赖关系。为了很好地组织起这样的复杂系统，能够按照计划执行，就必须要一个工作流调度系统来调度执行。例如，某个业务系统可能每天产生20G原始数据，我们每天都要对其进行处理。具体处理步骤如下所示：1. 通过Hadoop先将原...

白马啸西风 2022-09-06 18:25 评论0 收藏0
[译] 解密 Airbnb 的数据流编程神器：Airflow 中的技巧和陷阱

摘要：显然，这单独执行不起作用这将通过子操作符被作为像是自己的调度任务中那样运行。子也必须有个可用调度即使子作为其父的一部分被触发子也必须有一个调度如果他们的调度是设成，这个子操作符将不会触发任何任务。这两个例子都是缘起子操作符被当做了回填工作。 showImg(https://segmentfault.com/img/remote/1460000006768714); 前言 Airbnb的...

zsy888 2019-07-25 10:21 评论0 收藏0
[原]数据科学教程：如何使用Airflow调度数据科学工作流

摘要：概述是一个我们正在用的工作流调度器，相对于传统的任务管理，很好的为我们理清了复杂的任务依赖关系监控任务执行的情况。步骤三修改默认数据库找到配置文件修改配置注意到，之前使用的的方式是行不通的。微信号商业使用请联系作者。 showImg(https://segmentfault.com/img/remote/1460000006760428?w=1918&h=1556); 概述 Airfl...

v1 2019-07-25 10:35 评论0 收藏0
azkaban 安装

摘要：否则没有配置环境会报错。重新编译打包就可以 azkaban azkaban 是一个用Java开发的开源调度工具workflow,同类型的工具有用Python开发的 airflow 和 luigi. build 在我的windows上无法build成功，我在CentOS上build成功了 git clone https://github.com/azkaban/azkaban.git c...

luqiuwen 2019-08-15 12:21 评论0 收藏0