大数据必须框架-Azkaban

╰+哭是因爲堅強的太久メ 2023-10-04 16:44 34阅读 0赞

## 大数据必须框架-Azkaban ##

### 一、Azkaban 概论 ###

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

1）一个完整的数据分析系统通常都是由大量任务单元组成： Shell 脚本程序，Java 程序，MapReduce 程序、Hive 脚本等

2）各任务单元之间存在时间先后及前后依赖关系

3）为了很好地组织起这样的复杂执行计划，需要一个工作流调度系统来调度执行；  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70]

#### 2 、常见工作流调度系统 ####

1）简单的任务调度：直接使用 Linux 的 Crontab 来定义；

2）复杂的任务调度：开发调度平台或使用现成的开源调度系统，比如 Ooize、Azkaban、 Airflow、DolphinScheduler 等。

#### 3 、Azkaban 与 Oozie 对比 ####

总体来说，Ooize 相比 Azkaban 是一个重量级的任务调度系统，功能全面，但配置使用

也更复杂。 如果可以不在意某些功能的缺失， 轻量级调度器 Azkaban 是很不错的候选对象。

### 二、Azkaban 入门 ###

#### 1 、集群模式 安装 ####

##### 上传 tar 包 #####

**1）将 azkaban-db-3.84.4.tar.gz，azkaban-exec-server-3.84.4.tar.gz，azkaban-web-server-3.84.4.tar.gz 上传到 hadoop102 的/opt/software 路径**

$ ll

总用量 35572 
    -rw-r--r--. 1 atguigu atguigu     6433 4 月  18 17:24 azkaban-db-3.84.4.tar.gz 
    -rw-r--r--. 1 atguigu atguigu 16175002 4 月  18 17:26 azkaban-exec-server-3.84.4.tar.gz 
    -rw-r--r--. 1 atguigu atguigu 20239974 4 月  18 17:26 azkaban-web-server-3.84.4.tar.gz

**2）新建/opt/module/azkaban 目录，并将所有 tar 包解压到这个目录下**

$ mkdir /opt/module/azkaban

**3）解压 azkaban-db-3.84.4.tar.gz、 azkaban-exec-server-3.84.4.tar.gz 和 azkaban-web-server-3.84.4.tar.gz 到/opt/module/azkaban 目录下**

$ tar -zxvf azkaban-db-3.84.4.tar.gz -C /opt/module/azkaban/ 
    $  tar  -zxvf  azkaban-exec-server-3.84.4.tar.gz -C /opt/module/azkaban/ 
    $  tar  -zxvf  azkaban-web-server-3.84.4.tar.gz -C /opt/module/azkaban/

**4）进入到/opt/module/azkaban 目录，依次修改名称**

$  mv  azkaban-exec-server-3.84.4/ azkaban-exec 
    $  mv  azkaban-web-server-3.84.4/ azkaban-web

##### 配置 MySQL #####

**1）正常安装 MySQL**

**2）启动 MySQL**

$ mysql -uroot -p000000

**3）登陆 MySQL，创建 Azkaban 数据库**

mysql> create database azkaban;

\*\*4）创建 azkaban 用户并赋予权限 \*\*

**设置密码有效长度 4 位及以上**

mysql> set global validate_password_length=4;

**设置密码策略最低级别**

mysql> set global validate_password_policy=0;

**创建 Azkaban 用户，任何主机都可以访问 Azkaban，密码是 000000**

mysql> CREATE USER 'azkaban'@'%' IDENTIFIED BY '000000';

**赋予 Azkaban 用户增删改查权限**

mysql>  GRANT  SELECT,INSERT,UPDATE,DELETE  ON  azkaban.*  to 'azkaban'@'%' WITH GRANT OPTION;

**5）创建 Azkaban 表，完成后退出 MySQL**

mysql> use azkaban; 
    mysql>  source  /opt/module/azkaban/azkaban-db-3.84.4/create-all-sql-3.84.4.sql 
    mysql> quit;

**6）更改 MySQL 包大小；防止 Azkaban 连接 MySQL 阻塞**

$ sudo vim /etc/my.cnf

在\[mysqld\]下面加一行 max\_allowed\_packet=1024M

[mysqld] 
    max_allowed_packet=1024M

**8）重启 MySQL**

$ sudo systemctl restart mysqld

##### 配置 Executor Server #####

Azkaban Executor Server 处理工作流和作业的实际执行。  
**1）编辑 azkaban.properties**

$  vim  /opt/module/azkaban/azkaban-exec/conf/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=000000 
    mysql.numconnections=100

**2）同步 azkaban-exec 到所有节点**

$ xsync /opt/module/azkaban/azkaban-exec

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

$ bin/start-exec.sh 
    $ bin/start-exec.sh 
    $ bin/start-exec.sh

**注意：如果在/opt/module/azkaban/azkaban-exec 目录下出现 executor.port 文件，说明启动成功**

**4）下面激活 executor，需要**

$  curl  -G "hadoop102:12321/executor?action=activate" && echo 
    
    $  curl  -G "hadoop103:12321/executor?action=activate" && echo 
    
    $  curl  -G "hadoop104:12321/executor?action=activate" && echo

***如果三台机器都出现如下提示，则表示激活成功***

{
        "status":"success"}

##### 配置 Web Server #####

Azkaban Web Server 处理项目管理，身份验证，计划和执行触发。  
**1）编辑 azkaban.properties**

$  vim  /opt/module/azkaban/azkaban-web/conf/azkaban.properties

修改如下属性

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

**说明：**  
\#StaticRemainingFlowSize：正在排队的任务数；

\#CpuStatus：CPU 占用情况

\#MinimumFreeMemory ：内存占用情况 。 测试环境， 必须将 MinimumFreeMemory 删除掉， 否则 它会认为集群资源不够，不执行。

**2）修改 azkaban-users.xml 文件，添加 gis用户**

$ vim /opt/module/azkaban/azkaban-web/conf/azkaban-users.xml

<azkaban-users> 
      <user  groups="azkaban"  password="azkaban"  roles="admin" 
    username="azkaban"/> 
      <user password="metrics" roles="metrics" username="metrics"/> 
      <user password="gis" roles="admin" username="gis"/> 
      <role name="admin" permissions="ADMIN"/> 
      <role name="metrics" permissions="METRICS"/> 
    </azkaban-users>

**3）必须进入到 hadoop102 的/opt/module/azkaban/azkaban-web 路径，启动 web server**

$ bin/start-web.sh

4）访问 http://hadoop102:8081,并用 gis用户登陆

#### 2 、Work Flow 案例实操 ####

##### HelloWorld 案例 #####

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

azkaban-flow-version: 2.0

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

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

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

（1）Name：job 名称

（2）Type：job 类型。command 表示你要执行作业的方式为命令

（3）Config：job 配置

**3）将 azkaban.project、basic.flow 文件压缩到一个 zip 文件，文件名称必须是英文。**

**4）在 WebServer 新建项目：http://hadoop102:8081/index**

##### 作业依赖案例 #####

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

**具体步骤：**

**1）修改 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"

##### 自动 失败重试 案例 #####

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

**具体步骤：**  
**1）编译配置流**

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

**参数说明：**

retries：重试次数

retry.backoff：重试的时间间隔

### 三、Azkaban 进阶 ###

#### 1、JavaProcess 作业类型 案例 ####

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

Xms：最小堆

Xmx：最大堆

classpath：类路径

java.class：要运行的 Java 对象，其中必须包含 Main 方法

main.args：main 方法的参数

**案例：**

1）新建一个 azkaban 的 maven 工程

2）创建包名：com.gis

3）创建 AzTest 类

package com.gis; 
    
    public class AzTest {
         
        public static void main(String[] args) {
         
            System.out.println("This is for testing!"); 
        } 
    }

4）打包成 jar 包 azkaban-1.0-SNAPSHOT.jar

5）新建 testJava.flow，内容如下

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

#### 2、条件工作流 案例 ####

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

##### 运行时参数 案例 #####

**1）基本原理**

（1）父 Job 将参数写入 JOB\_OUTPUT\_PROP\_FILE 环境变量所指向的文件

（2）子 Job 使用 $\{jobName:param\}来获取父 Job 输出的参数并定义执行条件

**2）支持的条件运算符：**

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

**3）案例：**

需求：  
JobA 执行一个 shell 脚本。

JobB 执行一个 shell 脚本，但 JobB 不需要每天都执行，而只需要每个周一执行。

**新建 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

##### 预定义宏 案例 #####

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

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

（1）all_success: 表示父 Job 全部成功才执行(默认) 
    （2）all_done：表示父 Job 全部完成才执行
    （3）all_failed：表示父 Job 全部失败才执行 
    （4）one_success：表示父 Job 至少一个成功才执行 
    （5）one_failed：表示父 Job 至少一个失败才执行

**需求：**

JobA 执行一个 shell 脚本

JobB 执行一个 shell 脚本

JobC 执行一个 shell 脚本，要求 JobA、JobB 中有一个成功即可执行

**新建 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

#### 3、定时执行 案例 ####

**需求：JobA 每间隔 1 分钟执行一次；**

**Azkaban 可以定时执行工作流。在执行工作流时候，选择左下角 Schedule**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/63dfbdf334fb4f0887d6e5b013cba423.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/aff59e85579a47d5b78e287049072477.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70