Azkaban使用入门

接下来，记录一下azkaban工作流调度器的相关知识。一般一个分析系统，包含大量的分析单元，需要调用shell脚本、java程序、hive脚本等，这些脚本或者程序的调用会存在时间上的先后顺序，为了更好的组织管理这些任务，需要使用工作流调度器。比较常用的就是azkaban和oozie。除了使用第三方提供的调度器还可以自己开发，或使用linux自带的定时任务来实现。

azkaban基本知识

简介

azkaban是LinkedIn开源的批量工作流调度器，可以通过编写类似properties的文件，在里面定义工作的顺序和依赖关系。配置好azkaban后将可以通过web ui的方式，来管理和查看追踪工作流。官网https://azkaban.github.io，可以查看具体介绍，它提供了模块化的可插拔机制，支持command、java、hadoop和hive命令，另外它和任何版本的hadoop均兼容。

下图为官网展示的工作流形象图，定义了工作流后，在执行前可以通过图形化，更加直观的检查自己的工作流配置是否ok。

基本架构

如下图所示，azkaban主要由三部分组成，协同工作。

（1）azkaban web server：提供web ui，可以通过web实现对工作流的创建、执行、追踪等，安装后有单独的文件夹对应web server。

（2）azkaban executor server：负责工作流和任务的调度提交，安装后也有单独的文件夹对应executor server。

（3）mysql：解压对应的安装压缩文件后，需要执行相应脚本创建多张表，用于保存日志、执行计划和项目相关的信息。

运行模式

azkaban有三种运行模式。

（1）solo server mode：executor server和web server运行在同一个进程里，数据库使用内置的H2数据库，任务量不大的项目可以使用此模式。

（2）two server mode：executor server和web server运行在不同的进程里，数据库使用mysql。

（3）multiple executor mode： executor server和web server运行在不同的进程里，但是executor server可以有多个。

安装配置azkaban

安装配置，使用的是第二种模式two  server mode，执行服务器和web服务器分进程，暂都部署在同一个节点上，下面安装一下，安装之前需要提前安装hadoop、hive和mysql。

提前准备对应的安装包，这里选择的是cdh版本的hadoop。

# executor server安装包
-rwxr-xr-x 1 root root  11157302 Feb  2 17:31 azkaban-executor-server-2.5.0.tar.gz
# 导入mysql的脚本
-rwxr-xr-x 1 root root      1928 Feb  2 17:32 azkaban-sql-script-2.5.0.tar.gz
# web server安装包
-rwxr-xr-x 1 root root  11989669 Feb  2 17:32 azkaban-web-server-2.5.0.tar.gz

web server安装

（1）解压web server安装包到安装目录，需要提前准备好安装目录/kkb/install/azkaban。

[root@node01 /kkb/soft]# tar -zxvf azkaban-web-server-2.5.0.tar.gz -C /kkb/install/azkaban/

解压完成之后，修改安装目录名。

[root@node01 /kkb/install/azkaban]# ll
total 4
drwxr-xr-x 8 root root 4096 Feb  3 10:30 azkaban-web-2.5.0
# 重命名
[root@node01 /kkb/install/azkaban]# mv azkaban-web-2.5.0/ server
# 修改ok
[root@node01 /kkb/install/azkaban]# ll
total 4
drwxr-xr-x 8 root root 4096 Feb  3 10:30 server

（2）配置ssl安全访问协议，并将生成的keystore证书文件拷贝到上面重命名的server目录下。

# keytool：java数据证书的管理工具，使用户能够管理自己的公/私钥对及相关证书
# -keystore：指定密钥库的名称及位置(产生的各类信息将不在.keystore文件中)
# -alias：对我们生成的.keystore 进行指认别名；如果没有默认是mykey
# -genkey：在用户主目录中创建一个默认文件".keystore" 
# -keyalg：指定密钥的算法 RSA/DSA 默认是DSAga
[root@node01 /kkb/install/azkaban]# keytool -keystore keystore -alias jetty -genkey -keyalg RSA
# 输入密码，测试用123456
Enter keystore password:
Re-enter new password:
#随便写
What is your first and last name?
  [Unknown]:
What is the name of your organizational unit?
  [Unknown]:
What is the name of your organization?
  [Unknown]:
What is the name of your City or Locality?
  [Unknown]:
What is the name of your State or Province?
  [Unknown]:
# 输入两位的国家代码，中国为CN
What is the two-letter country code for this unit?
  [Unknown]:  CN
Is CN=Unknown, OU=Unknown, O=Unknown, L=Unknown, ST=Unknown, C=CN correct?
# 输入y确认
  [no]:  y
# 输入jetty密码，这里设置和keystore相同，直接回车即可
Enter key password for <jetty>
    (RETURN if same as keystore password):

Warning:
The JKS keystore uses a proprietary format. It is recommended to migrate to PKCS12 which is an industry standard format using "keytool -importkeystore -srckeystore keystore -destkeystore keystore -deststoretype pkcs12".
You have new mail in /var/spool/mail/root
# 在当前目录新生成keystore文件
[root@node01 /kkb/install/azkaban]# ll
total 8
-rw-r--r-- 1 root root 2231 Feb  3 10:35 keystore
drwxr-xr-x 8 root root 4096 Feb  3 10:30 server
# 拷贝到server
[root@node01 /kkb/install/azkaban]# cp keystore ./server/

（3）配置时区，配置为东八区，不配也可以，不影响运行。

# 交互式命令选择时区，选择亚洲，中国，北京时区即可
[hadoop@node01 /kkb/install/azkaban/server]$ tzselect
Please identify a location so that time zone rules can be set correctly.
Please select a continent or ocean.
 1) Africa
 2) Americas
 3) Antarctica
 4) Arctic Ocean
 5) Asia
 6) Atlantic Ocean
 7) Australia
 8) Europe
 9) Indian Ocean
10) Pacific Ocean
11) none - I want to specify the time zone using the Posix TZ format.
#? 5 
Please select a country.
 1) Afghanistan          18) Israel            35) Palestine
 2) Armenia          19) Japan            36) Philippines
 3) Azerbaijan          20) Jordan            37) Qatar
 4) Bahrain          21) Kazakhstan        38) Russia
 5) Bangladesh          22) Korea (North)        39) Saudi Arabia
 6) Bhutan          23) Korea (South)        40) Singapore
 7) Brunei          24) Kuwait            41) Sri Lanka
 8) Cambodia          25) Kyrgyzstan        42) Syria
 9) China          26) Laos            43) Taiwan
10) Cyprus          27) Lebanon            44) Tajikistan
11) East Timor          28) Macau            45) Thailand
12) Georgia          29) Malaysia            46) Turkmenistan
13) Hong Kong          30) Mongolia            47) United Arab Emirates
14) India          31) Myanmar (Burma)        48) Uzbekistan
15) Indonesia          32) Nepal            49) Vietnam
16) Iran          33) Oman            50) Yemen
17) Iraq          34) Pakistan
#? 9
Please select one of the following time zone regions.
1) Beijing Time
2) Xinjiang Time
#? 1

The following information has been given:

    China
    Beijing Time

Therefore TZ='Asia/Shanghai' will be used.
Local time is now:    Tue Feb  4 09:24:45 CST 2020.
Universal Time is now:    Tue Feb  4 01:24:45 UTC 2020.
Is the above information OK?
1) Yes
2) No
#? 1
# 复制生成的时区文件到本地覆盖
[hadoop@node01 /kkb/install/azkaban/server]$ cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime

（4）修改server/conf/azkaban.properties文件，注意注释不要写在配置后，否则启动会报错"Could not find or load main class #"。

#Azkaban Personalization Settings
azkaban.name=Test  #服务器ui的名字，用于服务器上方显示名字
azkaban.label=My Local Azkaban  #描述
azkaban.color=#FF3602 # ui颜色
azkaban.default.servlet.path=/index #默认servlet地址
web.resource.dir=web/  #默认根web目录
default.timezone.id=Asia/Shanghai  # 默认美国洛杉矶，修改为中国上海

#Azkaban UserManager class
user.manager.class=azkaban.user.XmlUserManager  # 用户权限管理默认类
user.manager.xml.file=conf/azkaban-users.xml  # 用户配置，里面添加admin用户

#Loader for projects
executor.global.properties=conf/global.properties  # 配置文件所在目录
azkaban.project.dir=projects # project目录

database.type=mysql # 数据库为mysql
mysql.port=3306
mysql.host=node01 # mysql安装在node01，选node01
mysql.database=azkaban
mysql.user=root
mysql.password=123456
#mysql.numconnections=100

# Velocity dev mode
velocity.dev.mode=false # jetty服务器属性

# Azkaban Jetty server properties.
jetty.maxThreads=25 # 线程最大数
jetty.ssl.port=8443 # ssl端口
jetty.port=8081 # jetty端口
jetty.keystore=keystore # ssl文件名
jetty.password=123456 # ssl密码
jetty.keypassword=123456 # jetty密码，和keystore相同
jetty.truststore=keystore # ssl文件名
jetty.trustpassword=123456 # ssl密码

# Azkaban Executor settings
executor.port=12321 # executor执行服务器端口

# mail settings
mail.sender=yangchaolin268@163.com # 发送邮箱
mail.host=stmp.163.com # 发送邮箱stmp地址
mail.user=yangchaolin268  #邮箱用户名
mail.password=*******   #邮箱用户密码
job.failure.email= #任务失败时发送邮箱
job.success.email= #任务成功时发送邮箱

lockdown.create.projects=false

cache.directory=cache

（5）conf/azkaban-user.xml中添加管理员用户，web登录时使用它来进入azkaban管理页面。

<azkaban-users>
    <user username="azkaban" password="azkaban" roles="admin" groups="azkaban" />
    <user username="metrics" password="metrics" roles="metrics"/>

    <role name="admin" permissions="ADMIN" />
    <role name="metrics" permissions="METRICS"/>

    <!--增加admin用户-->
    <user username="admin" password="admin" roles="admin,metrics" />
</azkaban-users>
You have new mail in /var/spool/mail/root

executor server安装

（1）解压executor server安装包到安装目录，安装目录就是上面已经安装了web server的根目录。

[root@node01 /kkb/soft]# tar -zxvf azkaban-executor-server-2.5.0.tar.gz -C /kkb/install/azkaban/

解压完成后，也修改安装目录名。

[root@node01 /kkb/install/azkaban]# ll
total 12
drwxr-xr-x 7 root root 4096 Feb  3 11:07 azkaban-executor-2.5.0
-rw-r--r-- 1 root root 2231 Feb  3 10:35 keystore
drwxr-xr-x 8 root root 4096 Feb  3 10:39 server
# 重命名
[root@node01 /kkb/install/azkaban]# mv azkaban-executor-2.5.0/ executor
[root@node01 /kkb/install/azkaban]# ll
total 12
drwxr-xr-x 7 root root 4096 Feb  3 11:07 executor
-rw-r--r-- 1 root root 2231 Feb  3 10:35 keystore
drwxr-xr-x 8 root root 4096 Feb  3 10:39 server

（2）修改executor/conf目录下的azkaban.properties文件，也需要修改时区，配置mysql连接。

#Azkaban
default.timezone.id=Asia/Shanghai

# Azkaban JobTypes Plugins
azkaban.jobtype.plugin.dir=plugins/jobtypes

#Loader for projects
executor.global.properties=conf/global.properties
azkaban.project.dir=projects

# 数据库修改为mysql，和sever类似
database.type=mysql
mysql.port=3306
mysql.host=node01
mysql.database=azkaban
mysql.user=root
mysql.password=123456
mysql.numconnections=100

# Azkaban Executor settings
executor.maxThreads=50
executor.port=12321
executor.flow.threads=30

mysql脚本导入

还有一个有关mysql脚本的包，也需要解压，然后执行里面的脚本初始化azkaban在mysql中的表。

（1）解压azkaban脚本sql压缩包。

[root@node01 /kkb/soft]# tar -zxvf azkaban-sql-script-2.5.0.tar.gz -C /kkb/install/azkaban/

解压完后，有很多脚本，需要选择如图所示的脚本，在mysql中导入，这样将批量创建表，单独创建的话就单个导入里面的其他脚本。

（2）mysql中导入脚本，按照前面web server和executor server里mysql database的配置，需提前创建azkaban数据库，切换到azkaban数据库后执行导入。

mysql> source /kkb/install/azkaban/azkaban-2.5.0/create-all-sql-2.5.0.sql

启动

先后启动web server/bin目录和executor server/bin目录下的启动脚本。

# 先后启动
[hadoop@node01 /kkb/install/azkaban/server]$ bin/azkaban-web-start.sh
# 另一个窗口启动
[hadoop@node01 /kkb/install/azkaban/executor]$ bin/azkaban-executor-start.sh

启动时没有报错，就可以通过https://node01:8443来访问web ui，登录用户名密码均为admin，访问时会报错不安全，可以忽略。登录ok后就显示如图所示界面，界面比较友好，创建、运行、追踪工作流的执行将比较方便。

azkaban使用 

接下来使用下azkaban，创建工作流。

（1）command类型，多个job工作流

 可以新建以job为后缀的文件，在里面写命令，写两个job，job1执行完就执行job2，job2的执行需依赖job1先执行完。

job1，文件名job1.job

# 第一个job
type=command
command=echo 'this is command 1'

job2，文件名job2.job，依赖job1，dependencies写第一个job的文件名，如果有多个依赖，需要用逗号隔开。

# 第二个job
type=command
# 如果有多个依赖，使用逗号隔开
dependencies=job1
command=echo 'this is command 2 buddy'

编写完后，将两个文件打成zip包。接下来就需要使用web ui创建project，然后将zip包上传上去，正常就会显示如下界面，可以看出job2依赖job1。

点击Execute Flow后，就跳入执行确认界面。两个job的依赖关系也能直观展示，点击Execute就可以执行job，另外如果是要设置定时任务，就点击左下角的schedule。

执行完成后，可以通过web查看到job1执行完后才执行job2，执行的起始时间和结束时间都有，通过detail可以查看具体的执行细节。

点击job2的detail查看细节，可以看到正常的打印了job2中的内容。

 

以上就是使用azkaban执行工作流的基本步骤，即创建project→上传编写好的文件并打包上传→执行工作流(或者设置定时任务执行)→查看工作流执行情况。

（2）执行hdfs任务，可以写一个job，里面执行一段hdfs命令。

job内容，上传一个本地文件到hdfs，并提前创建好目录。

# 运行关于hdfs相关的任务
type=command
command=echo "start hdfs job"
command.1=/kkb/install/hadoop-2.6.0-cdh5.14.2/bin/hadoop fs -mkdir /azkaban
command.2=/kkb/install/hadoop-2.6.0-cdh5.14.2/bin/hadoop fs -put /home/hadoop/readme.txt /azkaban
command.3=echo "hdfs job finished!"

经历创建项目和文件打包上传后，执行工作流，通过detail查看日志，发现成功执行。

 继续查看在hdfs新建了azkaban目录，并将本地文件readme.txt上传到了目录。

[hadoop@node01 ~/azkaban/out]$ hadoop fs -ls /azkaban
20/02/04 12:45:42 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Found 1 items
-rw-r--r--   3 hadoop supergroup         36 2020-02-03 15:27 /azkaban/readme.txt
# 查看内容
[hadoop@node01 ~/azkaban/out]$ hadoop fs -cat /azkaban/readme.txt
20/02/04 12:45:55 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
i come from changsha what about you

（3）执行hive任务，实现在默认hive数据库中，创建表，加载本地数据，查看数据和写出数据到本地。

首先先写一段hive脚本

use default;
--建表
create table if not exists test_azkaban(id int,name string,city string) row format delimited fields terminated by ',';
--加载数据
load data local inpath '/home/hadoop/azkaban/test.txt' overwrite into table test_azkaban;
--将查询数据，插入新表
create table if not exists addressgroup as select city,count(*) from test_azkaban group by city;
--将addressgroup表中数据导出到本地
insert overwrite local directory '/home/hadoop/azkaban/out' select * from addressgroup;    

数据准备，给test.txt中加数据。

1,clyang,changsha
2,messi,changsha
3,ronald,shanghai
4,herry,shanghai

再编写一个job，里面写执行上面的脚本。

# 执行hive job
type=command
command=echo 'start execute hive job'
command.1=/kkb/install/hive-1.1.0-cdh5.14.2/bin/hive -f 'hive.sql'
command.2=echo 'hive job finished!'

经历创建项目和文件打包上传后，执行工作流，通过detail查看日志，发现成功执行。

查看输出目录，发现多了out目录，继续查看内容，已经成功执行了hive脚本，统计出结果。

（4）执行mapreduce任务，刚开始安装hadoop时，hadoop jar命令使用自带的jar包执行了单词次数统计，也可以写一个工作流来实现。

job内容如下，其中wordcount是jar包里要执行的类名，并传递两个参数，分别为要统计的文本路径/readme.txt和统计输出路径/wordcount/output。

# 执行一个MapReduce job
type=command
command=echo "start mapreduce job"
command.1=/kkb/install/hadoop-2.6.0-cdh5.14.2/bin/hadoop jar hadoop-mapreduce-examples-2.6.0-cdh5.14.2.jar wordcount /readme.txt /wordcount/output
command.2=echo "mapreduce job finished!"

经历创建项目，和文件打包（hadoop的这个jar包也一起打包）上传后，执行工作流，通过detail查看日志，发现成功执行。

查看mr统计结果，ok。

（5）执行定时任务，除了使用linux的自带的crontab，也可以使用azkaban。

job内容如下。

#scheduler.job
type=command
command= echo "hello buddy"

经历创建项目和文件打包上传后，进入工作流确认界面后，需要选择Schedule来设置定时规则。

测试选择执行频率1分钟一次，查看历史记录可以看出每隔1分钟执行了定时任务。

（6）web给工作流传递参数，可以通过job文件指定接受web ui传递的参数。

job内容如下，使用${参数名}来接收，参数名需和web ui中的参数名一致。

# 可以通过webui传递参数
type=command
name=${inputname}
age=${inputage}
command=echo ${name} ${age}

经历创建项目、打包上传zip文件，启动工作流进入确认画面，选择Flow Parameters设置参数。

参数设置界面。

执行后，ok，正确打印出姓名和年龄。

以上，就是azkaban使用入门，记录一下后面使用。