Azkaban从入门到精通以及案例实操系列

1、Azkaban概论

1.1、Azkaban简介

Azkaban 是一个开源的基于 Web 的工作流调度系统，由 LinkedIn 公司开发并维护。它可以帮助用户在大规模数据处理中来管理和调度作业，提供了简单易用、高效可靠的工作流设计和调度功能。

Azkaban 的主要特点包括：

丰富的用户界面：提供了基于 Web 的交互式用户界面，使得用户可以方便地创建、编辑和执行工作流，同时也提供了丰富的监控和报警功能。
灵活的工作流定义：使用 Azkaban 可以根据需求灵活地定制工作流，支持各种类型的任务（如 Hadoop MapReduce、Spark、Hive、Pig、Shell、Java 等）以及任务之间的依赖、并行和串行执行等功能。
高效的任务调度：Azkaban 提供了高效的任务调度功能，可以根据任务优先级、资源限制等情况智能地分配任务资源，实现高效的任务调度和运行。
安全可靠：Azkaban 支持多用户和多组权限控制，并提供了各种安全措施（如 SSL/TLS 加密、Kerberos 认证等）来确保数据安全和隐私保护。

1.2、为什么需要工作流调度系统

例如，我们可能有这样一个需求，某个业务系统每天产生20G原始数据，我们每天都要对其进行处理，处理步骤如下所示：

通过Hadoop先将原始数据上传到HDFS上（HDFS的操作）；
使用MapReduce对原始数据进行清洗（MapReduce的操作）；
将清洗后的数据导入到hive表中（hive的导入操作）；
对Hive中多个表的数据进行JOIN处理，得到一张hive的明细表（创建中间表）；
通过对明细表的统计和分析，得到结果报表信息（hive的查询操作）；

在这里插入图片描述

要完成这个业务系统需要考虑的角度：

一个完整的数据分析系统通常都是由大量任务单元组成：shell脚本程序，java程序，mapreduce程序、hive脚本等。
各任务单元之间存在时间先后及前后依赖关系。
为了很好地组织起这样的复杂执行计划，需要一个工作流调度系统来调度执行。

像上述需求，特别是在一些涉及多个复杂任务、数据依赖关系复杂、任务之间存在关联和交互等方面，手工调度和管理变得困难且低效。因此，为了保证任务的正常执行，以及简化任务调度管理就显得尤为重要，这也就是需要工作流调度系统的原因。

1.3、常见工作流调度系统

简单的任务调度：直接使用Linux的Crontab来定义；
复杂的任务调度：开发调度平台或使用现成的开源调度系统，比如Ooize、Azkaban、 Airflow、DolphinScheduler等。

2、Azkaban入门

2.1、Azkaban的部署方式

Azkaban 可以通过以下两种方式进行部署：

单机部署：将 Azkaban 作为一个 Java 应用程序，通过在一台服务器上运行一个或多个 Azkaban Web 服务器和 Azkaban Executor 服务器来实现。这种部署方式适合用于学习，或者小型数据处理场景，可以快速部署，但面对大规模数据处理时可能会存在性能瓶颈。
分布式部署：将 Azkaban 部署到多个服务器上，可以通过将 Azkaban Web 服务器和 Azkaban Executor 服务器分别分配到不同的服务器上来实现高可用、高性能的工作流调度系统。这种部署方式适合大型数据处理场景，具有更好的扩展性和容错能力。

总之，选择何种部署方式主要取决于实际场景和需求。对于小规模或试验性应用，单机部署可能是更加简便的选择；而对于大规模的生产环境，则需要采用分布式部署以满足高可用、高性能、高扩展性等要求。本文主要使用集群部署。

2.2、Azkaban的集群部署

2.2.1、软件下载

下载地址

2.2.2、软件上传

在这里插入图片描述

2.2.3、解压软件，修改名字

[song@hadoop102 software]$ mkdir /opt/model/azkaban
[song@hadoop102 software]$ tar -zxvf azkaban-db-3.84.4.tar.gz -C/opt/model/azkaban/[song@hadoop102 software]$ tar -zxvf azkaban-exec-server-3.84.4.tar.gz -C/opt/model/azkaban/[song@hadoop102 software]$ tar -zxvf azkaban-web-server-3.84.4.tar.gz -C/opt/model/azkaban/[song@hadoop102 azkaban]$ mv azkaban-exec-server-3.84.4/ azkaban-exec
[song@hadoop102 azkaban]$ mv azkaban-web-server-3.84.4/ azkaban-web

在这里插入图片描述

2.2.4、配置MySQL

2.2.4.1、正常安装即可，电脑上已经安装了，这就不多介绍了

2.2.4.2、登陆MySQL，创建Azkaban数据库

#创建Azkaban用户，任何主机都可以访问Azkaban，密码是azkaban
CREATE USER 'azkaban'@'%' IDENTIFIED BY 'azkaban';

在这里插入图片描述

2.2.4.3、创建azkaban用户并赋予权限

#赋予azkaban用户增删改查权限 
GRANT SELECT,INSERT,UPDATE,DELETE ON azkaban.* to'azkaban'@'%' WITH GRANT OPTION;

在这里插入图片描述

2.2.4.4、更改MySQL包大小；防止Azkaban连接MySQL阻塞

在这里插入图片描述

在[mysqld]下面加一行max_allowed_packet=1024M

[mysqld]
max_allowed_packet=1024M

在这里插入图片描述

2.2.4.5、重启MySQL

sudo systemctl restart mysqld

2.2.4.6、加载SQL

mysql> source /opt/model/azkaban/azkaban-db-3.84.4/create-all-sql-3.84.4.sql

在这里插入图片描述

2.2.5、配置Executor Server

Executor ServerWebhadoop102，hadoop103，hadoop104hadoop102

编辑azkaban.properties

在这里插入图片描述

[song@hadoop102 conf]$ vim azkaban.properties

修改内容

#...default.timezone.id=Asia/Shanghai
#...
azkaban.webserver.url=http://hadoop102:8081

executor.port=12321
#...
database.type=mysql
mysql.port=3306
mysql.host=hadoop102
mysql.database=azkaban
mysql.user=azkaban
mysql.password=azkaban
...

同步azkaban-exec到所有节点

在这里插入图片描述

必须进入到/opt/model/azkaban/azkaban-exec路径，分别在三台机器上，启动executor server

bin/start-exec.sh

在这里插入图片描述
注意：如果在

/opt/model/azkaban/azkaban-exec

目录下出现executor.port文件，说明启动成功
在这里插入图片描述

激活executor

[song@hadoop102 azkaban-exec]$ curl -G"hadoop102:12321/executor?action=activate"&& echo
[song@hadoop103 azkaban-exec]$ curl -G"hadoop103:12321/executor?action=activate"&& echo
[song@hadoop104 azkaban-exec]$ curl -G"hadoop104:12321/executor?action=activate"&& echo

如果三台机器都出现

{"status":"success"}

提示，则表示激活成功
在这里插入图片描述

2.2.6、配置Web Server

Azkaban Web Server处理项目管理，身份验证，计划和执行触发。

编辑azkaban-web下的azkaban.properties

vim azkaban.properties

修改内容为

...default.timezone.id=Asia/Shanghai...
database.type=mysql
mysql.port=3306
mysql.host=hadoop102
mysql.database=azkaban
mysql.user=azkaban
mysql.password=azkaban
mysql.numconnections=100...
azkaban.executorselector.filters=StaticRemainingFlowSize,CpuStatus

StaticRemainingFlowSize：正在排队的任务数；
CpuStatus：CPU占用情况
MinimumFreeMemory 过滤器会检查 executor 主机空余内存是否会大于 6G，如果不足 6G，则 web-server 不会将任务交由该主机执行
azkaban.executorselector.filters=StaticRemainingFlowSize,CpuStatus

修改azkaban-users.xml文件，添加song用户

在这里插入图片描述

必须进入到hadoop102的/opt/model/azkaban/azkaban-web路径，启动web server

[song@hadoop102 azkaban-web]$ bin/start-web.sh

在这里插入图片描述

访问http://hadoop102:8081,并用song用户登陆![在这里插入图片描述](https://img-blog.csdnimg.cn/20cf3850de6649a7936e2c11dd41eb28.png)

在这里插入图片描述

3、案例实操

3.1、HelloWorld案例

在windows环境，新建azkaban.project文件，编辑内容如下

azkaban-flow-version:2.0

注意：该文件作用，是采用新的Flow-API方式解析flow文件。

新建basic.flow文件，内容如下

nodes:- name: jobA
    type: command
    config:
      command: echo "Hello World"

Name：job名称
Type：job类型。command表示你要执行作业的方式为命令
Config：job配置

将azkaban.project、basic.flow文件压缩到一个zip文件，文件名称必须是英文。
在WebServer新建项目：hadoop102:8081/index
给项目名称命名和添加项目描述
文件上传
执行任务流

在这里插入图片描述

在日志中，查看运行结果

在这里插入图片描述

3.2、任务作业依赖案例

需求：JobA和JobB执行完了，才能执行JobC

修改basic.flow为如下内容

nodes:- name: jobC
    type: command
    # jobC 依赖 JobA和JobB
    dependsOn:- jobA
      - jobB
    config:
      command: echo "I’m JobC"- name: jobA
    type: command
    config:
      command: echo "I’m JobA"- name: jobB
    type: command
    config:
      command: echo "I’m JobB"

将修改后的basic.flow和azkaban.project压缩成second.zip文件
重复HelloWorld中的后续步骤

3.3、自动失败重试案例

需求：如果执行任务失败，需要重试3次，重试的时间间隔10000ms

编译配置流

nodes:- name:JobA
    type: command
    config:
      command: sh /not_exists.sh
      retries:3
      retry.backoff:10000

参数说明：

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

将修改后的basic.flow和azkaban.project压缩成four.zip文件
重复HelloWorld后续步骤,步骤略。
执行并观察到一次失败+三次重试

也可以在Flow全局配置中添加任务失败重试配置，此时重试配置会应用到所有Job。
案例如下：

config:
  retries:3
  retry.backoff:10000
nodes:- name:JobA
    type: command
    config:
      command: sh /not_exists.sh

3.4、手动失败重试案例

需求：JobA=》JobB（依赖于A）=》JobC=》JobD=》JobE=》JobF。生产环境，任何Job都有可能挂掉，可以根据需求执行想要执行的Job。
具体步骤：

编译配置流

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: echo "This is JobC."- 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."

将修改后的basic.flow和azkaban.project压缩成five.zip文件
重复HelloWorld后续步骤。

在这里插入图片描述

Enable和Disable下面都分别有如下参数：

Parents：该作业的上一个任务
Ancestors：该作业前的所有任务
Children：该作业后的一个任务
Descendents：该作业后的所有任务
Enable All：所有的任务

可以根据需求选择性执行对应的任务。

在这里插入图片描述

3.5、JavaProcess作业类型案例

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

Xms：最小堆
Xmx：最大堆
classpath：类路径
java.class：要运行的Java对象，其中必须包含Main方法
main.args：main方法的参数

案例：

新建一个azkaban的maven工程
创建包名：com.song
创建AzTest类

packagecom.song;publicclassAzTest{publicstaticvoidmain(String[] args){System.out.println("This is for testing!");}}

打包成jar包test_demo-1.0-SNAPSHOT.jar
新建testJava.flow，内容如下

nodes:- name: test_java
    type: javaprocess
    config:Xms:96M
      Xmx:200M
      java.class:com.song.AzTest

将Jar包、flow文件和project文件打包成javatest.zip
创建项目=》上传javatest.zip =》执行作业=》观察结果

在这里插入图片描述

3.6、条件工作流案例

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

基本原理 - 父Job将参数写入JOB_OUTPUT_PROP_FILE环境变量所指向的文件- 子Job使用 ${jobName:param}来获取父Job输出的参数并定义执行条件
支持的条件运算符：

（1）==    等于
（2）!=    不等于
（3）>    大于
（4）>=    大于等于
（5）<    小于
（6）<=    小于等于
（7）&& 与
（8）||    或
（9）!    非

3.6.1、运行时参数案例

需求：JobA执行一个shell脚本，JobB执行一个shell脚本，但JobB不需要每天都执行，而只需要每个周一执行。
新建JobA.sh

#!/bin/bash
echo "do JobA"
wk=`date +%w`
echo "{\"wk\":$wk}"> $JOB_OUTPUT_PROP_FILE

新建JobB.sh

#!/bin/bash
echo "do JobB"

4 新建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打包成condition.zip
创建condition项目=》上传condition.zip文件=》执行作业=》观察结果
按照我们设定的条件，JobB会根据当日日期决定是否执行。

3.6.2、预定义宏案例

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/bash
echo "do JobA"

新建JobC.sh

#!/bin/bash
echo "do JobC"

新建macro.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

JobA.sh、JobC.sh、macro.flow、azkaban.project文件，打包成macro.zip。注意：没有JobB.sh。
创建macro项目=》上传macro.zip文件=》执行作业=》观察结果

在这里插入图片描述

3.7、定时执行案例

需求：JobA每间隔1分钟执行一次；
具体步骤：

Azkaban可以定时执行工作流。在执行工作流时候，选择左下角Schedule
右上角注意时区是上海，然后在左面填写具体执行事件，填写的方法和crontab配置定时任务规则一致。

在这里插入图片描述

3.8、邮件报警案例

3.8.1、注册邮箱

申请注册一个126邮箱
点击邮箱账号=》账号管理
开启SMTP服务
一定要记住授权码

3.8.2、默认邮件报警案例

Azkaban默认支持通过邮件对失败的任务进行报警，配置方法如下：
1）在azkaban-web节点hadoop102上，编辑

/opt/model/azkaban/azkaban-web/conf/azkaban.properties

，修改如下内容：

[song@hadoop102 azkaban-web]$ vim /opt/model/azkaban/azkaban-web/conf/azkaban.properties

添加如下内容：

#这里设置邮件发送服务器，需要 申请邮箱，切开通stmp服务，以下只是例子
[email protected]
mail.host=smtp.126.com
[email protected]
mail.password=用邮箱的授权码

保存并重启web-server。

[song@hadoop102 azkaban-web]$ bin/shutdown-web.sh
[song@hadoop102 azkaban-web]$ bin/start-web.sh

编辑basic.flow

nodes:- name: jobA
    type: command
    config:
      command: echo "This is an email test."

将azkaban.project和basic.flow压缩成email.zip
创建工程=》上传文件=》执行作业=》查看结果
观察邮箱，发现执行成功或者失败的邮件

3.9、电话报警案例

3.9.1、第三方告警平台集成

有时任务执行失败后邮件报警接收不及时，因此可能需要其他报警方式，比如电话报警。如有类似需求，可与第三方告警平台进行集成，例如睿象云。

进入睿象云官网注册账号并登录,官网地址：https://www.aiops.com/

在这里插入图片描述

集成告警平台，使用Email集成

在这里插入图片描述

获取邮箱地址，后边需将报警信息发送至该邮箱

在这里插入图片描述

配置分派策略

在这里插入图片描述

配置通知策略

在这里插入图片描述

测试，执行上一个邮件通知的案例，将通知对象改为刚刚集成第三方平台时获取的邮箱。

在这里插入图片描述

3.10、Azkaban多Executor模式注意事项

Azkaban多Executor模式是指，在集群中多个节点部署Executor。在这种模式下， Azkaban web Server会根据策略，选取其中一个Executor去执行任务。
为确保所选的Executor能够准确的执行任务，我们须在以下两种方案任选其一，推荐使用方案二。

3.10.1、方案一：指定特定的Executor（hadoop102）去执行任务。

在MySQL中azkaban数据库executors表中，查询hadoop102上的Executor的id。

mysql> use azkaban;Reading table information for completion of table and column names
You can turn off this feature toget a quicker startup with-ADatabase changed
mysql> select * from executors;+----+-----------+-------+--------+| id | host          | port  | active |+----+-----------+-------+--------+|1| hadoop103 |35985|1||2| hadoop104 |36363|1||3| hadoop102 |12321|1|+----+-----------+-------+--------+3 rows in set (0.00 sec)

在执行工作流程时加入useExecutor属性，如下

在这里插入图片描述

3.10.2、方案二：在Executor所在所有节点部署任务所需脚本和应用。

标签： hadoop 大数据 Azkaban

本文转载自: https://blog.csdn.net/prefect_start/article/details/130828710
版权归原作者 后端技术那点事 所有，如有侵权，请联系我们删除。

Azkaban从入门到精通以及案例实操系列

1、Azkaban概论

1.1、Azkaban简介

1.2、为什么需要工作流调度系统

1.3、常见工作流调度系统

2、Azkaban入门

2.1、Azkaban的部署方式

2.2、Azkaban的集群部署

2.2.1、软件下载

2.2.2、软件上传

2.2.3、解压软件，修改名字

2.2.4、配置MySQL

2.2.4.1、正常安装即可，电脑上已经安装了，这就不多介绍了

2.2.4.2、登陆MySQL，创建Azkaban数据库

2.2.4.3、创建azkaban用户并赋予权限

2.2.4.4、更改MySQL包大小；防止Azkaban连接MySQL阻塞

2.2.4.5、重启MySQL

2.2.4.6、加载SQL

2.2.5、配置Executor Server

2.2.6、配置Web Server

3、案例实操

3.1、HelloWorld案例

3.2、任务作业依赖案例

3.3、自动失败重试案例

3.4、手动失败重试案例

3.5、JavaProcess作业类型案例

3.6、条件工作流案例

3.6.1、运行时参数案例

3.6.2、预定义宏案例

3.7、定时执行案例

3.8、邮件报警案例

3.8.1、注册邮箱

3.8.2、默认邮件报警案例

3.9、电话报警案例

3.9.1、第三方告警平台集成

3.10、Azkaban多Executor模式注意事项

3.10.1、方案一：指定特定的Executor（hadoop102）去执行任务。

3.10.2、方案二：在Executor所在所有节点部署任务所需脚本和应用。

发表评论

“Azkaban从入门到精通以及案例实操系列”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航