当前位置： 首页 > news >正文

Azkaban 深度探索：Flow 2.0、报警机制与脚本优化

news 2025/5/11 1:24:05

        在大数据时代，数据处理和任务调度变得越来越复杂。为了高效地管理和执行各种任务，我们需要强大的工具来进行任务编排和调度。Azkaban 就是这样一个强大的任务调度和编排工具，它可以帮助我们轻松地管理和执行各种复杂的任务。本文将深入探讨 Azkaban 的 Flow 2.0 用法、报警机制以及脚本优化问题，并通过实际案例和代码演示来帮助读者更好地理解和应用 Azkaban。

一、Azkaban 简介

（一）什么是 Azkaban

Azkaban 是一个开源的工作流调度系统，用于在分布式环境中管理和执行复杂的任务流。它提供了一个可视化的界面，方便用户创建、调度和监控任务流。Azkaban 支持多种任务类型，包括 shell 脚本、Java 程序、Hive 脚本、Spark 任务等。

（二）Azkaban 的特点

  1. 简单易用：Azkaban 提供了一个直观的可视化界面，方便用户创建、调度和监控任务流。用户可以通过拖拽的方式创建任务流，无需编写复杂的代码。

  2. 强大的任务调度功能：Azkaban 支持多种任务调度方式，包括定时调度、依赖调度等。用户可以根据自己的需求设置任务的执行时间和依赖关系，确保任务按照正确的顺序执行。

  3. 高可靠性：Azkaban 采用了分布式架构，具有高可靠性和容错性。即使某个节点出现故障，任务也能够继续执行，不会影响整个系统的运行。

  4. 易于扩展：Azkaban 提供了丰富的插件机制，用户可以根据自己的需求扩展系统的功能。例如，用户可以开发自己的任务类型插件，实现对特定任务的支持。

二、Flow 2.0 的用法

（一）小试牛刀

创建文件

需要创建两个文件，分别是JobA.flow和JobA.project。

在JobA.project文件中编写一句话。

azkaban-flow-version: 2.0

在JobA.flow中，编写任务，需要注意YAML语句。

nodes:- name: jobAtype: commandconfig:command: echo "this is a simple test"

YAML 语句注意事项

YAML 是一种简洁的数据序列化格式，书写格式要求严格，层级关系纯靠缩进进行表示。

在编写任务时，需要注意缩进的正确性，以确保任务的结构清晰。

打包文件

将JobA.flow和JobA.project两个文件打包成一个 zip 文件，以便上传到 Azkaban 进行任务执行。

（二）YAML 格式的数据

YAML 与 JSON、XML 的比较

JSON 和 XML 是常见的数据格式，都是用来存储和传输数据的。而 YAML 则是近年来越来越流行的一种数据格式。

YAML 与 JSON 和 XML 一样，都可以用于传递信息，但它具有更加简洁和易读的语法。
JSON  存储数据，比如 datax中大量使用了json

        "reader": {"name": "mysqlreader","parameter": {"username": "root","password": "123456","column": ["id","age"],"splitPk": "id","connection": [{"table": ["user"],"jdbcUrl": ["jdbc:mysql://192.168.52.12:3306/sqoop"]}]}}

编写好了一个json ，里面含有大量的信息，肯定是有人用了。用的人肯定解析了这个文件。
JSON 传输数据：   html --> Java(SSM SSH SpringBoot)
xml 也可以做到。但是慢慢的淘汰了。
这三类文件格式都可以充当配置文件（存储数据）
JSON:  core.json
XML ： hdfs-site.xml
YAML : 即将开始 

例如，在 Spring Boot 中，支持application.properties和application.yaml两种配置文件，YAML 格式的配置文件更加易于阅读和维护。

YAML 语句注意事项：

1. 大小写敏感
2. 使用缩进表示层级关系 ；
3. 缩进长度没有限制，只要元素对齐就表示这些元素属于一个层级；
4. 使用#表示注释 ；
5. 字符串默认不用加单双引号，但单引号和双引号都可以使用，双引号表示不需要对特殊字符进行转义；
6. YAML 中提供了多种常量结构，包括：整数，浮点数，字符串，NULL，日期，布尔，时间。
7、缩进不允许使用tab，只允许空格

YAML 的语法格式

YAML 的语法格式非常严格，层级关系纯靠缩进进行表示。

表示对象

# value 与 ： 符号之间必须要有一个空格

key: value

map的写法

# 写法一 同一缩进的所有键值对属于一个map

key:

key1: value1

key2: value2

# 写法二

{key1: value1, key2: value2}

数组的表示方式

# 写法一 使用一个短横线加一个空格代表一个数组项

- a

- b

- c

# 写法二

[a,b,c]

关于单双引号

s1: '内容\n 字符串'

s2: "内容\n 字符串"

转换后：

{ s1: '内容\\n 字符串', s2: "内容\n 字符串" }

如果你想表示一个换行：

s1: "内容\n 字符串"

Flow 2.0 建议将公共参数定义在 `config` 下，并通过 `${}` 进行引用。

（三）多任务依赖

复习安装

在使用 Azkaban 进行任务调度时，我们需要先进行安装和配置。

安装过程包括上传安装包、解压、生成 MySQL 元数据、配置 web-server 和 exec-server 等步骤。

Azkaban：大数据任务调度与编排工具的安装与使用-CSDN博客

多任务编排

Azkaban 可以同时执行多个任务，也可以单独执行某个任务。

多任务编排靠dependsOn这个属性来实现任务之间的依赖关系。

many.flow

nodes:- name: jobEtype: commandconfig:command: echo "This is job E"# jobE depends on jobDdependsOn:- jobD- name: jobDtype: commandconfig:command: echo "This is job D"# jobD depends on jobA、jobB、jobCdependsOn:- jobA- jobB- jobC- name: jobAtype: commandconfig:command: echo "This is job A"- name: jobBtype: commandconfig:command: echo "This is job B"- name: jobCtype: commandconfig:command: echo "This is job C"

many.project

azkaban-flow-version: 2.0

上传运行

可以一次执行多个任务，也可以单独执行某个任务

定时任务

Azkaban 还支持定时任务的设置，可以在特定的时间自动执行任务。

定时任务的设置可以通过在任务的 YAML 定义中添加schedule属性来实现。

实战：将我们刚才的多任务的job,定时执行：

（四）内嵌流（嵌套流）案例

不带参数传递的内嵌流

内嵌流是 Azkaban Flow 2.0 中的一个强大功能，它允许我们在一个任务流中嵌套另一个任务流。

不带参数传递的内嵌流使用相对简单，只需要在任务的 YAML 定义中使用flow类型，并指定内嵌流的文件路径即可。

例如，以下是一个不带参数传递的内嵌流的 YAML 示例：

nodes:- name: jobCtype: commandconfig:command: echo "This is job C"dependsOn:- embedded_flow- name: embedded_flowtype: flownodes:- name: jobBtype: commandconfig:command: echo "This is job B"dependsOn:- jobA- name: jobAtype: commandconfig:command: echo "This is job A"

带有参数的内嵌流

有时候，我们需要在内嵌流中传递参数，以便在不同的情况下执行不同的任务。

带有参数的内嵌流可以通过在任务的 YAML 定义中使用parameters属性来实现。

例如，以下是一个带有参数的内嵌流的 YAML 示例：

neiqianflow.flow

nodes:- name: jobCtype: commandconfig:command: echo "This is job C"dependsOn:- embedded_flow- name: embedded_flowtype: flowconfig:name: zhangsannodes:- name: jobBtype: commandconfig:command: echo "This is job B ${name}"dependsOn:- jobA- name: jobAtype: commandconfig:command: echo "This is job A"

neiqianflow.project

azkaban-flow-version: 2.0

（五）动态传参

代码演示

在 Azkaban 的任务中，我们可以使用动态传参来根据不同的情况传递不同的参数。

以下是一个动态传参的代码演示：

nodes:- name: jobCtype: commandconfig:command: echo "This is job C"dependsOn:- test_flow- name: test_flow#此处是type=flowtype: flowconfig:prop: ${canshu}#此处有nodesnodes:- name: jobBtype: commandconfig:command: echo "This is job B ${prop}"dependsOn:- jobA- name: jobAtype: commandconfig:command: echo "This is job A"

打包上传

完成动态传参的代码编写后，我们需要将包含动态传参代码的文件打包成 zip 文件，并上传到 Azkaban 进行任务执行。

三、Azkaban 的报警机制

（一）邮箱通知

邮箱服务器的选择

要实现邮箱通知，我们需要有一个邮箱服务器。

常见的邮箱服务器有 Gmail、163、126 等。

我们可以根据自己的需求选择合适的邮箱服务器，并注册一个邮箱账号。

授权码的获取

以 163 邮箱为例，我们需要获取授权码才能使用邮箱服务器发送邮件通知。

授权码的获取方法可以在邮箱的设置中找到，一般需要开启 SMTP 服务，并按照提示生成授权码。

配置邮箱通知

在 Azkaban 的 web-server 端，我们需要进行邮箱配置，以便能够使用邮箱服务器发送邮件通知。

配置方法包括在azkaban.properties文件中设置邮箱相关的参数，如mail.sender、mail.host、mail.user、mail.password等。

例如，以下是一个邮箱配置的示例：

mail.sender=your-email@163.com
mail.host=smtp.163.com
mail.user=your-email@163.com
mail.password=your-authorization-code

配置完成后，我们需要重启 web-server 服务，使配置生效。

设置接收邮箱地址

在执行任务时，我们可以设置接收邮件通知的邮箱地址。

可以在任务的 YAML 定义中使用job.failure.email和job.success.email属性来设置任务失败和成功时的接收邮箱地址。

例如，以下是一个设置接收邮箱地址的 YAML 示例：

jobs:- name: task1type: commandcommand: echo "Task 1"job.failure.email: recipient1@example.comjob.success.email: recipient2@example.com

（二）电话报警机制

使用第三方平台

除了邮箱通知，Azkaban 还可以通过第三方平台实现电话报警机制。

例如，我们可以使用https://www.aiops.com/这个平台来生成一个 appKey，并将其配置到 Azkaban 中，实现电话报警功能。

（三）Azkaban 的重试机制

重试机制的作用

在任务执行过程中，可能会出现各种问题导致任务失败。

为了提高任务的可靠性，Azkaban 提供了重试机制。

当任务失败时，Azkaban 会自动尝试重新执行任务，直到任务成功或者达到最大重试次数。

配置重试机制

可以在任务的 YAML 定义中使用retries属性来设置任务的重试次数。

例如，以下是一个设置重试次数的 YAML 示例：

# 重试次数
retries: 3
# 每次重试时间间隔
retry.backoff: 10000案例：
nodes:- name: jobAtype: commandconfig:command: echo "this is a simple test"retries: 3retry.backoff: 10000

在这个示例中，task1将在失败时最多重试 3 次。

四、Azkaban 脚本优化的问题

目前遇到的问题：
1、启动azkaban的时候，需要启动两个甚至多个脚本,比较麻烦
    start-web.sh  start-exec.sh
2、日志查看没有放在一块，需要切换目录比较麻烦

优化方案：
编写一个脚本  azkaban.sh

（一）源码分析

配置脚本

在 Azkaban 的脚本中，我们可以通过配置脚本来优化任务的执行。

配置脚本可以包括设置环境变量、加载外部配置文件等。

例如，以下是一个配置脚本的示例：

#!/bin/bash
# 参数个数一个或者两个 not equal to --> ne
if  [ $# -lt 1 ] && [ $# -gt 2 ] 
then
echo -e  "\033[32mUsage1:\033[0m azkaban.sh 1 \033[32mor\033[0m azkaban.sh 0"
echo -e  "\033[32mUsage2:\033[0m azkaban.sh 1 w \033[32mor\033[0m azkaban.sh 0 w"
exit;
fi# 输入 1 为启动, 输入 0 为关闭
v1="web"
v2="exec"if [ $1 -eq 1 ]
then
stat="start-"
elif [ $1 -eq 0 ]
then
stat="shutdown-"
else
echo -e "\033[32mUsage:\033[0m azkaban.sh 1 \033[32mor\033[0m azkaban.sh 0"
exit;
ficd /opt/installs/azkaban/logs
# 传入第二个参数 w 代表 web-server , e 代表 exec-server 不传第二个默认全部启动/停止
if [ "$2" = 'w' ]
then
path1=/opt/installs/azkaban/${v1}-server/bin/${stat}${v1}.sh
echo -e "\033[32m${stat}${v1}.sh\033[0m"
sh  ${path1}
elif [ "$2" = 'e' ]
then
path2=/opt/installs/azkaban/${v2}-server/bin/${stat}${v2}.sh
echo -e "\033[32m${stat}${v2}.sh\033[0m"
sh  ${path2}
elif [ 'a'"$2"'a' = 'aa' ]
then
path1=/opt/installs/azkaban/${v1}-server/bin/${stat}${v1}.sh
echo -e "\033[32m${stat}${v1}.sh\033[0m"
sh  ${path1}
sleep 2
path2=/opt/installs/azkaban/${v2}-server/bin/${stat}${v2}.sh
echo -e "\033[32m${stat}${v2}.sh\033[0m"
sh  ${path2}
else
echo -e "\033[32mUsage1:\033[0m azkaban.sh 1 w \033[32mor\033[0m azkaban.sh 0 w"
echo -e "\033[32mUsage2:\033[0m azkaban.sh 1 e \033[32mor\033[0m azkaban.sh 0 e"
exit;
fisleep 2
echo -e "\033[32m-------------------jps--------------------\033[0m"
jps

赋予权限

cd /usr/local/bin
将脚本放入这个目录chmod 777 azkaban.sh

创建 logs 目录

在执行任务时，我们可以创建一个 logs 目录来存储任务的日志文件。

这样可以方便我们查看任务的执行情况和排查问题。

cd /opt/installs/azkaban/
用于存储两个服务的日志
mkdir logs

使用

# 使用方法
# 1.启动azkaban
azkaban.sh 1# 单独启动Web
azkaban.sh 1 w# 单独启动exec
azkaban.sh 1 e# 2.关闭azkaban
azkaban.sh 0# 单独关闭Web
azkaban.sh 0 w# 单独关闭exec
azkaban.sh 0 e

五、实际案例：使用 Java 代码编写一个任务，给别人发邮件

（一）导入包

导入所需的包

在使用 Java 代码发送邮件时，我们需要导入以下几个包：

<dependency><groupId>com.sun.mail</groupId><artifactId>javax.mail</artifactId><version>1.6.2</version></dependency><dependency><groupId>javax.mail</groupId><artifactId>javax.mail-api</artifactId><version>1.4.7</version></dependency>

（二）编写代码

代码实现

以下是一个使用 Java 代码发送邮件的示例：

package com.email;/**** @Date(时间)2023-05-12* @Author xx**** 发送邮件*/
public class Email_Code {//测试一下public static void main(String[] args) throws Exception {for (int i = 0; i < 10; i++) {EmailUtils.sendEmail("18638147931@163.com","2844408713@qq.com","赛哥","六哥","某某大厦301房间等你","暗号：三长两短");}}}

（三）工具类代码

工具类的作用

为了方便在 Azkaban 任务中使用发送邮件的功能，我们可以将发送邮件的代码封装成一个工具类。

这样可以在需要发送邮件的地方直接调用工具类的方法，提高代码的可维护性和可复用性。

工具类的实现

以下是一个发送邮件的工具类代码：

package com.email;import javax.mail.Session;
import javax.mail.Transport;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;
import java.util.Date;
import java.util.Properties;public class EmailUtils {public static void sendEmail(String fromAccount,String toAccount,String fromName,String toName,String title,String content) throws Exception {// 参数配置，⽤于连接邮件服务器Properties props = new Properties();// 使⽤协议props.setProperty("mail.transport.protocol", "smtp");// 发件⼈邮箱的 SMTP 服务器地址props.setProperty("mail.smtp.host", "smtp.163.com");// 需要请求认证props.setProperty("mail.smtp.auth", "true");// 创建会话对象，⽤于与邮箱服务器交互Session session = Session.getInstance(props);// 设置为debug模式，在控制台中可以查看详细的发送⽇志session.setDebug(true);// 1.创建邮件对象MimeMessage message = new MimeMessage(session);// 2.设置发件⼈，其中 InternetAddress 有三个参数，分别为：邮箱，显示的昵称，昵称的字符集编码message.setFrom(new InternetAddress(fromAccount, fromName, "UTF-8"));// 3.设置收件⼈ - MimeMessage.RecipientType.TOmessage.setRecipient(MimeMessage.RecipientType.TO, new InternetAddress(toAccount ,toName, "UTF-8"));// 7.设置邮件主题message.setSubject(title,"UTF-8");// 8.设置邮件正⽂内容，指定格式为HTML格式message.setContent(content, "text/html;charset=UTF-8");// 9.设置显示发件时间message.setSentDate(new Date());// 10.保存设置message.saveChanges();// 根据 Session 获取邮件传输对象Transport transport = session.getTransport();// 连接邮件服务器transport.connect("18638147931@163.com", "MAGBDQDGKEHCBVQA");// 发送邮件transport.sendMessage(message, message.getAllRecipients());// 关闭连接transport.close();}
}

六、总结

        Azkaban 是一个功能强大的任务调度和编排工具，通过 Flow 2.0 的用法，我们可以更加灵活地管理和执行任务。多任务依赖和定时任务的设置使得任务的执行更加高效和自动化。内嵌流和动态传参的功能为任务的编排提供了更多的可能性。

        在报警机制方面，邮箱通知和电话报警机制可以及时通知用户任务的执行情况，提高系统的可靠性。重试机制则可以在任务执行失败时自动进行重试，减少人工干预的成本。

        对于脚本优化问题，合理的配置脚本和创建 logs 目录可以提高脚本的可读性和可维护性。使用 Java 代码编写发送邮件的任务，可以方便地实现任务执行结果的通知功能。


http://www.mrgr.cn/news/41724.html

相关文章：

  • leetcode每日一题day22(24.10.2)——准时到达的列车最小时速
  • 网站开发基础：JavaScript
  • 哪些芯片封装需要基板，哪些不需要？
  • 如何从 PC 中检索已删除的文件？从 PC 恢复已删除的照片技巧
  • 解决OpenCV保存视频 视频全部为绿色的bug
  • string的实现（下）
  • 生信初学者教程（二十三）：REF+SVM筛选候选标记物
  • 大语言模型入门（三）——提示词编写注意事项
  • MYSQL 乐观锁
  • 路由交换实验指南
  • [产品管理-41]：什么是变量、属性、特征、特性？他们的相同点、不同点？
  • C++ 语言特性10 - 委托构造函数
  • 开放式耳机哪个品牌好？分享几款不错的开放式蓝牙耳机
  • C++常引用详解
  • Java开发类加载器实现机制
  • Visual Studio 使用技巧之界面布局
  • 浏览器 F12 application 应用程序面板
  • 10. 模块
  • 大学城就餐推荐系统小程序的设计
  • 深入理解 Git 一个开发者的必备工具