大数据hadoop之——任务调度器azkaban（azkaban环境部署）（hadoop任务调度平台）-ag凯发k8国际

大数据平台技术框架支持的开发语言多种多样，开发人员的背景差异也很大，这就产生出很多不同类型的程序（任务）运行在大数据平台之上，如：mapreduce、hive、pig、spark、java、shell、python 等。

这些任务需要不同的运行环境，并且除了定时运行，各种类型之间的任务存在依赖关系，一张简单的任务依赖图如下：

• crontab （linux 自带命令，使用方式简单，适合不是非常复杂的场景，比如只按照时间来调度）
• oozie（ hadoop 自带的开源调度系统，使用方式比较复杂，适合大型项目场景）
• azkaban（一个开源调度系统，使用方式比较简单，适合中小型项目场景）
• 企业定制开发（企业自研的调度系统，不开源）

azkaban是由linkedin开源的一个批量工作流任务调度器，azkaban来自linkedin公司，用于管理他们的hadoop批处理工作流。日常生产环境中，为了得到想要的数据，通常需要执行很多作业，一批作业执行完毕，再将中间结果进一步处理，最后得到有价值的数据，因此作业之间执行有先后顺序和依赖关系。这样的一组作业称为一个工作流，azkaban就是用来构建、运行和管理工作流的工具，它提供友好的web用户界面来维护和跟踪用户的工作流程。azkaban已经在linkedin运行了好几年，管理着很多hadoop和数据仓库作业流，具有很强的可用性。

官方文档：
https://azkaban.readthedocs.io/en/latest/
https://azkaban.github.io/azkaban/docs/latest/

简单来讲，它有几个特点：

• 分布式多执行器
• mysql重试
• 友好的用户界面
• 有条件的工作流
• 数据触发
• 高安全
• 支持插件扩展，从web ui到作业执行
• 全作者身份管理系统

azkaban 和 oozie 是市面上最流行的两种调度器。总体来说，ooize 相比 azkaban 是一个重量级的任务调度系统，功能全面，但部署和使用也更复杂，比较适合作为大型项目的任务调度系统。而 azkaban 相对而言，配置和使用更为简单，能够满足常见的任务调度，比较适合作为中小型项目的任务调度系统。

对比如下：

azkaban 三大核心组件

• 关系型元数据库（mysql）
• azkaban web server
• azkaban executor server

azkabanwebserver 是 azkaban 的主要管理者，负责项目管理、身份验证、调度和监控执行,并且为用户界面

提交和执行工作流，记录工作流日志，和 azkaban webserver 可以在同一台服务器，也可部署在独立的机器。把 executor 单独分开有几个好处：

• 在多 executor 模式下可以方便扩展
• 工作流在某一个 executor 挂掉，可以在另一个 executor 上重试
• 可以滚动升级，从而不影响调度

负责存储azkaban系统的数据，包括用户上传的工作流文件、作业执行的日志等。executor server和web server都通过jdbc频繁地对其操作。

azkaban 元数据库

在3.0版本之后，我们提供了两种模式：独立的“单独服务器”模式和分布式多执行器模式。下面介绍两种模式的区别。

• solo server mode(单机模式)：该模式中web server和executor server运行在同一个进程中，进程名azkabansingleserver。可以使用自带的h2数据库或者配置mysql数据。该模式适用于小规模的使用。
• multiple executor mode（多执行器模式），适用于大规模的生产环境。它的数据库应该由设置了主从关系的mysql实例进行备份。web server和executor server运行在不同的进程中，这样升级和维护就不会影响到用户。这种多主机设置为阿兹卡班带来了健壮和可伸缩的方面。

目前我们采用的是multiple executor mode方式，分别在不同的主机上部署多个azkaban executorserver以应对高并发定时任务执行的情况，从而减轻单个服务器的压力。

安装步骤如下：

• 设置数据库
• 配置数据库以使用多个执行程序
• 为数据库中配置的每个executor下载并安装executor server
• azkaban安装插件
• 安装web服务器

官方文档：https://azkaban.readthedocs.io/en/latest/getstarted.html

$ cd /opt/bigdata/hadoop/software$ git clone https://github.com/azkaban/azkaban.git

$ cd azkaban; ./gradlew build installdist

【温馨提示】如果编译失败了，就多执行几次

$ cd azkaban-solo-server/build/install/azkaban-solo-server$ ./bin/start-solo.sh$ netstat -tnlp|grep 8081### 停止服务，这里不执行$ ./bin/shutdown-solo.sh

检查进程

$ jps

• 访问http://ip:8081 (默认端口是8081)
• zkaban默认登录名/密码：azkaban/azkaban

访问：http://192.168.0.113:8081

阿兹卡班独立服务器默认不使用ssl。但你也可以在独立的web服务器上用同样的方法设置它。azkaban web服务器支持ssl套接字连接器，这意味着必须提供密钥存储库，您可以按照以下步骤生成这里提供的有效的jetty密钥存储库:

创建ssl配置

$ keytool -keystore keystore -alias jetty -genkey -keyalg rsa1.输入密钥库口令: 1234562.再次输入新口令: 1234563.[unknown]: azkaban4.[unknown]: azkaban5.[unknown]: azkaban6.[unknown]: shenzhen7.[unknown]: guangdong8.[unknown]: cn9.[no]: y 10.(如果和密钥库口令相同, 按回车):

修改azkaban 配置文件azkaban.properties或azkaban.private.properties(推荐)。

$ cd /opt/bigdata/hadoop/software/azkaban/azkaban-solo-server/build/install/azkaban-solo-server/conf$ touch azkaban.private.properties

在azkaban.private.properties（文件需要创建）配置如下：

# 根据上面设置的填，keystore文件会自动生成# ssl 文件名jetty.keystore=/opt/bigdata/hadoop/software/azkaban/azkaban-solo-server/build/install/azkaban-solo-server/keystorejetty.password=123456jetty.keypassword=123456# 文件名jetty.truststore=/opt/bigdata/hadoop/software/azkaban/azkaban-solo-server/build/install/azkaban-solo-server/keystorejetty.trustpassword=123456

在azkaban.properties修改如下配置：

jetty.use.ssl=truejetty.ssl.port=8443

重启服务

$ ./bin/shutdown-solo.sh ; ./bin/start-solo.sh$ jps$ netstat -tnlp|grep 8443

web访问验证

https://192.168.0.113:8443

先停掉上面的服务

$ /opt/bigdata/hadoop/software/azkaban/azkaban-solo-server/build/install/azkaban-solo-server/bin/shutdown-solo.sh

部署规划

这里也提供一下我编译的安装包，下载地址如下：

链接：https://pan.baidu.com/s/1zvuyfxg3scpqbfeswy-olq
提取码：6666

$ mkdir /opt/bigdata/hadoop/server/azkaban$ cd /opt/bigdata/hadoop/software/azkaban/$ cp ./azkaban-web-server/build/distributions/azkaban-web-server-3.91.0-313-gadb56414.tar.gz /opt/bigdata/hadoop/server/azkaban/$ cp ./azkaban-exec-server/build/distributions/azkaban-exec-server-3.91.0-313-gadb56414.tar.gz /opt/bigdata/hadoop/server/azkaban/$ cp ./azkaban-db/build/distributions/azkaban-db-3.91.0-313-gadb56414.tar.gz /opt/bigdata/hadoop/server/azkaban/### 解压并改名$ cd /opt/bigdata/hadoop/server/azkaban

因为我之前安装过了mysql，不清楚的可以参考我之前的文章：

#【温馨提示】一般公司禁止mysql -u root -p123456这种方式连接，在history里有记录，存在安全隐患，小伙伴不要被公司安全审计哦，切记！！！$ mysql -u root -p输入密码：123456create database azkaban;create user 'azkaban'@'%' identified by 'azkaban';grant select,insert,update,delete on azkaban.* to 'azkaban'@'%' with grant option;

mysql数据包大小可能需要重新配置。默认情况下，mysql允许的包大小可能低得离谱。要增加它，您需要将属性max_allowed_packet设置为一个更大的数字，比如1024m。要在linux中配置，请打开/etc/my.cnf或者/etc/my.cnf.d/mysql-server.cnf（推荐），在mysqld后面的某个地方，添加以下内容:

[mysqld]max_allowed_packet=1024m

重启mysql服务

$ systemctl restart mysqld$ netstat -tnlp|grep 3306

开始初始化azkaban表

$ cd /opt/bigdata/hadoop/server/azkaban/azkaban-db# 连接mysql$ mysql -u root -p密码：123456use azkaban# 可能版本不一样，sql文件也不太一样，create-all-sql-*.sqlsource create-all-sql-3.91.0-313-gadb56414.sql

$ cd /opt/bigdata/hadoop/server/azkaban/azkaban-exec# mysql配置，如果不一样，就需要调整$ grep mysql conf/azkaban.properties

修改conf/azkaban.properties配置文件

### 修改时区default.timezone.id=asia/shanghai### 修改mysql hostmysql.host=hadoop-node1### webserver.urlazkaban.webserver.url=https://hadoop-node1:8443### executor.port不设置就是随机值了，不方便管理，所以这里还是固定一个端口号，看资料大部分都是使用12321这个端口，这里也随大流executor.port=12321

启动服务

# 【温馨提示】必须进入到azkaban-exec目录下执行启动重启命令，因为配置文件中有些路径用的是相对路径$ cd /opt/bigdata/hadoop/server/azkaban/azkaban-exec# 重启$ ./bin/shutdown-exec.sh ; ./bin/start-exec.sh# azkabanexecutorserver$ jps$ telnet -tnlp|grep 12321

在数据库中查看记录

通过接口的方式去激活，不能直接改表字段值，切记！！！

# 记得换成自己的ip或域名$ curl -g "hadoop-node1:12321/executor?action=activate" && echo

【温馨提示】重启azkaban executor server得重新激活

• 【第一步】先登录hadoop-node2创建azkaban目录

$ mkdir -p /opt/bigdata/hadoop/server/azkaban

• 【第二步】登录到hadoop-node1 copy 安装目录到hadoop-node2

$ cd /opt/bigdata/hadoop/server/azkaban$ scp -r azkaban-exec hadoop-node2:/opt/bigdata/hadoop/server/azkaban/

• 【第三步】启动executor server

# 登录到hadoop-node2 切换到azkaban目录$ cd /opt/bigdata/hadoop/server/azkaban/azkaban-exec$ ./bin/start-exec.sh$ jps$ netstat -tnlp|grep 12321

• 【第四步】激活executor server

# 记得换成自己的ip或域名$ curl -g "hadoop-node2:12321/executor?action=activate" && echo

【温馨提示】重启azkaban executor server得重新激活

$ cd /opt/bigdata/hadoop/server/azkaban/azkaban-web# mysql配置，如果不一样，就需要调整$ grep mysql conf/azkaban.properties

修改conf/azkaban.properties配置文件

### 修改时区default.timezone.id=asia/shanghai### 修改mysql hostmysql.host=hadoop-node1### azkaban.executorselector.filters调度策略# 把minimumfreememory去掉，因为minimumfreememory是6g，自己电脑资源有限，如果小伙伴的机器资源雄厚，可以保留# staticremainingflowsize：根据排队的任务数来调度任务到哪台executor机器# cpustatus：跟据cpu空闲状态来调度任务到哪台executor机器azkaban.executorselector.filters=staticremainingflowsize,cpustatus

启动服务

$ ./bin/start-web.sh$ jps$ netstat -tnlp|grep 8081

跟上面的一样

$ keytool -keystore keystore -alias jetty -genkey -keyalg rsa1.输入密钥库口令: 1234562.再次输入新口令: 1234563.[unknown]: azkaban4.[unknown]: azkaban5.[unknown]: azkaban6.[unknown]: shenzhen7.[unknown]: guangdong8.[unknown]: cn9.[no]: y 10.(如果和密钥库口令相同, 按回车):

配置

在azkaban-web/azkaban.private.properties（文件需要创建）配置如下：

# 根据上面设置的填，keystore文件会自动生成# ssl 文件名jetty.keystore=/opt/bigdata/hadoop/server/azkaban/azkaban-web/keystorejetty.password=123456jetty.keypassword=123456# 文件名jetty.truststore=/opt/bigdata/hadoop/server/azkaban/azkaban-web/keystorejetty.trustpassword=123456

在azkaban-web/azkaban.properties修改如下配置：

jetty.use.ssl=truejetty.ssl.port=8443

修改azkaban-exec/conf/azkaban.properties

jetty.port=8443# where the azkaban web server is locatedazkaban.webserver.url=https://hadoop-node1:8443

重启服务

$ ./bin/shutdown-web.sh ; ./bin/start-web.sh$ jps$ netstat -tnlp|grep 8443

web访问验证：https://192.168.0.113:8443

官方文档：https://azkaban.readthedocs.io/en/latest/usermanager.html

$ cd /opt/bigdata/hadoop/server/azkaban/azkaban-web$ cat conf/azkaban-users.xml

配置一个管理员用户，增加如下一行

重启web服务

$ ./bin/shutdown-web.sh ; ./bin/start-web.sh

https://hadoop-node1:8443

官方文档：https://azkaban.readthedocs.io/en/latest/createflows.html

1、在windows环境，新建helloworld.project文件，编辑内容如下：

azkaban-flow-version: 2.0

【温馨提示】该文件作用，是采用新的flow-api方式解析flow文件，内容基本上是固定的，2.0版本xxx.flow是yaml格式，2.0之前的版本是key=value格式，示例如下：

type=commandcommand=echo 'hello'

2、新建helloworld.flow文件，内容如下：

nodes: - name: joba type: command config: command: echo "hello world"

【温馨提示】注意缩进的空格

• name：job名称
• type：job类型。command表示你要执行作业的方式为命令
• config：job配置
  3、将上面两个文件压缩成一个zip文件，目前只支持zip文件，文件名称必须是英文。

4、新建project

5、把zip文件上传到azkaban执行

6、执行

• dependentworkflow.flow

nodes: - name: joba type: command config: command: echo "joba" - name: jobb type: command config: command: echo "jobb" - name: jobc type: command dependson: - joba - jobb config: command: echo "jobc"

• dependentworkflow.project

azkaban-flow-version: 2.0

• 创建project，并把zip文件上传到azkaban执行

• 执行（执行完joba和jobb才执行jobc）

执行一个不存在的脚本/tmp/retry.sh，则任务失败，间隔10000ms，重试3次，其实总共会执行4次，1 3（重试3次）

• autofailed2retry.flow

---nodes: - name: joba type: command config: command: sh /tmp/retry.sh retries: 3 retry.backoff: 10000

retries：重试次数
retry.backoff：重试的间隔时间（ms）

• autofailed2retry.project

azkaban-flow-version: 2.0

• 创建project，并把zip文件上传到azkaban执行

• 执行

执行了4次，最后一次执行的状态为失败状态

需求：joba=》jobb（依赖于a）=》jobc（依赖于b）=》jobd（依赖于c）=》jobe（依赖于d）=》jobf（依赖于e）。生产环境，任何job都可能挂掉，可以根据需求执行想要执行的job。这里假设jobc失败了。

• manualfailed2retry.flow

---nodes: - name: joba type: command config: command: echo "this is joba." - name: jobb type: command dependson: - joba config: command: echo "this is jobb." - name: jobc type: command dependson: - jobb config: command: sh /tmp/retry.sh - name: jobd type: command dependson: - jobc config: command: echo "this is jobd." - name: jobe type: command dependson: - jobd config: command: echo "this is jobe." - name: jobf type: command dependson: - jobe config: command: echo "this is jobf."

• manualfailed2retry.project

azkaban-flow-version: 2.0

• 创建project，并把zip文件上传到azkaban执行

• 执行

执行到jobc失败了，后面的job就会自动取消了

• 手动创建这个/tmp/retry.sh脚本，每个executor都创建这个脚本，因为不确定会调度到哪个executor

$ echo "echo 'this is jobc.'" > /tmp/retry.sh

可以看到，之前joba和jobb执行成功的就不再执行了。正是预期效果。

【温馨提示】type不单单只有command，还有javaprocess，当然还有其它type，例如：noop等。可以参考官方文档
【概述】

javaprocess类型可以运行一个自定义主类方法，type类型为javaprocess，可用的配置为：

• xms：最小堆内存
• xmx：最大堆内存
• classpath：类路径，可以省略，省略的话，是flow当前文件路径
• java.class：要运行的java对象，其中必须包含main方法
• main.args：main方法的参数
• 新建azkaban的maven工程或者module
• 创建com.bigdata.aztest类，内容如下：

【示例】

package com.bigdata;public class aztest { public static void main(string[] args) { system.out.println("this is azkaban test!!!"); }}

• 打包成jar包azkaban-1.0-snapshot.jar

• 新建com.bigdata.testjava.flow，内容如下：

nodes: - name: az_javaprocess_test type: javaprocess config: xms: 100m xmx: 200m java.class: com.bigdata.aztest

• javaprocesstest001.project，project文件是固定的也是并不可少的。

azkaban-flow-version: 2.0

• 把三个文件打包成zip包

• 把zip包上传到azkaban上执行

条件工作流功能允许用户自定义执行条件来决定是否运行某些job，条件可以由当前job的父job输出的运行时参数构成，也可以使用预定义宏。在这些条件下，用户可以在确定job执行逻辑时获取得更大的灵活性，例如：只要父job之一成功，就可以运行当前job。

• 父job将参数写入job_output_prop_file环境变量所指向的文件
• 子job使用${jobname.param}来获取父job输出的参数并定义执行条件

• == 等于
• != 不等于
• 大于
• = 大于等于
• < 小于
• <= 小于等于
• && 与
• || 或
• ! 非

【示例一】
需求：joba执行一个shell脚本。job执行一个shell脚本，但jobb不需要每天都执行，而只需要每周一执行。

• 新建joba.sh

#!/bin/bashecho "do joba"wk=`date %w`echo "{"wk":$wk}" > $job_output_prop_file

获取当前周第几天，0：周日，1表示周一，则jobb需要到周一才执行，今天不执行

• 新建jobb.sh

#!/bin/bashecho "do job"

• 新建condition.flow

nodes: - name: joba type: command config: command: sh joba.sh - name: jobb type: command dependson: - joba config: command: sh jobb.sh condition: ${joba:wk} == 1

• 将joba.sh、jobb.sh、condition.flow和azkaban.project打包成xxx.zip

• 创建condition项目=》上传xxx.zip文件=》执行作业=》观察结果

azkaban中预置了几个特殊的判断条件，称为预定于宏。

预定于宏会根据所有父job的完成情况进行判断，再决定是否执行。可用的预定义宏如下：

• all_success：表示父job全部成功才执行（默认）
• all_done：表示父job全部完成才执行
• all_failed：表示父job全部失败才执行
• one_success：表示父job至少一个成功才执行
• one_failed：表示父job至少一个失败才执行

需求：

• joba执行一个shell脚本
• jobb执行一个shell脚本
• jobc执行一个shell脚本，要求joba、jobb中有一个成功即可执行

步骤如下：

• 新建joba.sh

#!/bin/bashecho "do joba"

• 新建jobb.sh（不创建，验证）

#!/bin/bashecho "do jobb"

• 新建jobc.sh

#!/bin/bashecho "do jobc"

• 新建marco.flow

nodes: - name: joba type: command config: command: sh joba.sh - name: jobb type: command config: command: sh jobb.sh - name: jobc type: command dependson: - joba - jobb config: command: sh jobc.sh condition: one_success

• 新建marco.project

azkaban-flow-version: 2.0

• 把文件打包成zip文件，并上传到azkaban上执行

• 执行

jobb执行失败了，jobc还是执行成功，验证成功。

拿之前的案例，设置定时调度

每分钟执行一次

把定时任务移除

这里以qq邮箱为示范，需要开启邮箱协议

$ cd /opt/bigdata/hadoop/server/azkaban/azkaban-web$ vi conf/azkaban.properties# 修改的内容如下：mail.sender=2920992033@qq.commail.host=smtp.qq.com# 下面两行没有需要增加mail.user=2920992033@qq.com# password就换成上面截图的授权码（自己邮箱授权码）mail.password=xxxx

重启web服务生效

$ ./bin/shutdown-web.sh ; ./bin/start-web.sh

还是拿上面的案例进行测试

查看邮箱，发现已经收到了工作流报警邮件

因为azkaban默认是不支持电话报警的，所以这里使用睿象云做中转实现电话报警。

这里选择免费试用，自己测试就免费试用也够了，如果是商用或者在企业里用肯定是需要付费使用的。

使用智能告警平台

选择集成

这里使用email集成方式，因为azkaban支持邮件

配置这个邮箱：881841810azkaban@camail.aiops.com

能收到告警电话，验证ok。小伙伴可以自己试试。

总结

原生的 azkaban 支持的plugin类型有以下这些：

• command：linux shell命令行任务
• javaprocess：原生java任务
• gobblin：通用数据采集工具
• hadoopjava：运行hadoopmr任务
• hive：支持执行hivesql
• pig：pig脚本任务
• spark：spark任务
• hdfstoteradata：把数据从hdfs导入teradata
• teradatatohdfs：把数据从teradata导入hdfs

上面我们示例中用到了command和javaprocess，其中最简单而且最常用的是command类型。

azkaban基础部分就先到这了，后续会有更多相关的文章，请小伙伴耐心等待~