工作流调度 Azkaban 工作流 Flow2.0入门

Azkaban使用

azkaban 4.x目前同时支持flow1.0与flow 2.0；
官网说flow 1.0将来会被淘汰，所以本文档使用flow 2.0
如果对flow 1.0感兴趣的同学，可以参考文章自行学习体验
Azkaba内置的任务类型支持command、java

1. Flow 2.0

1. 入门例子Hello World

在windows或mac中，创建文件flow20.project，内容如下

1	azkaban-flow-version: 2.0

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

nodes:

- name: jobA

type: command

config:

command: echo "Hello World! This is an flow 2.0 example."

文件中要有nodes的部分，包含所有你想要运行的job

需要为每个job指定name、type

大多的job需要config

将两个文件压缩成一个zip包，比如起名叫Archive.zip
使用如下账号信息登录azkaban

1 2	用户名：kkbrwe 密码：kkb123

创建工程

上传项目zip文件

执行flow

弹出如下界面

下图Execution queued successfully with exec id 8表示，web server选择id是8的exec server执行此流

绿色表示执行成功；查看Job List

查看日志

2. 单job有多个command

flow中的一个job有多个command

nodes:

- name: jobA

type: command

config:

command: mkdir /export/servers/azkaban-exec-server-4.0.0/executions/test1

command.1: mkdir /export/servers/azkaban-exec-server-4.0.0/executions/test2

第一个command用command表示

第二个用command.1表示

第三个用command.2表示，以此类推

剩余步骤（图略）
- 生成项目zip文件
- web ui创建项目
- 上传zip文件
- 执行flow
- 查看日志

3. 包含多个有依赖关系job的flow

job间可以相互依赖，创建flow文件dependon.flow，内容如下

nodes:

- name: jobC

type: noop

# jobC depends on jobA and jobB

dependsOn:

- jobA

- jobB

- name: jobA

type: command

config:

command: echo "This is echoed by jobA."

- name: jobB

type: command

config:

command: pwd

jobC依赖jobA、jobB

Noop: A job that takes no parameters and is essentially a null operation. Used for organizing your graph

以下操作跟上边的例子入门例子Hello World相似
dependon.flow与flow20.project压缩生成zip文件dependon.zip
web server ui界面创建项目，然后上传项目zip文件，然后执行，并查看Job List及job日志

以下是jobA日志

以下是jobB日志

4. 自动失败重试案例

如果job执行失败，可以配置成自动重试若干次，每次重试时间间隔一定时长

nodes:

- name: JobA

type: command

config:

command: sh /a_non_exists.sh

retries: 3

retry.backoff: 3000

说明：

/a_non_exists.sh是一个不存在的sh脚本

retries重试次数

retry.backoff每次重试的时间间隔，单位毫秒

以下操作跟上边的例子入门例子Hello World相似
retry.flow与flow20.project压缩生成zip文件retry.zip
web server ui界面创建项目，然后上传项目zip文件，然后执行，并查看Job List及job日志
部分截图略

日志

也可以在flow文件中，加入全局重试次数，此重试配置对flow文件中的所有job都生效；内容如下

config:

retries: 3

retry.backoff: 3000

nodes:

- name: JobA

type: command

config:

command: sh /a_non_exists.sh

5. 手动失败重试案例

手动失败重试场景：
- 对于某些flow中的失败job，不能通过自动重试解决的，比如并非一些系统短时的问题，比如暂时的网络故障导致的超时、暂时的资源不足导致的执行失败
- 此时需要手动的做些处理后，然后再进行重新执行flow中job
- 跳过成功的job
- 从失败的job开始执行
一个flow中，有5个job，有依赖关系如下
- jobE依赖jobD；
- jobD依赖jobC；
- jobC依赖jobB
- jobB依赖jobA
创建flow文件manulretry.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 /export/servers/azkaban-exec-server-4.0.0/tmp.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."

压缩生成zip包、web ui创建项目、上传zip、执行flow

查看jobC的日志

原因是没有找到下图的sh脚本文件

那么在对应的exec server的对应目录下创建此sh脚本文件

由于不确定，flow重试时，web选择哪个exec执行flow，所以保险起见，有两个方法
方法一：在3个exec节点中都创建tmp.sh脚本

[hadoop@node01 azkaban-exec-server-4.0.0]$ cd

[hadoop@node01 ~]$ cd /export/servers/azkaban-exec-server-4.0.0

[hadoop@node01 azkaban-exec-server-4.0.0]$ vim tmp.sh

[hadoop@node02 azkaban-exec-server-4.0.0]$ cd

[hadoop@node02 ~]$ cd /export/servers/azkaban-exec-server-4.0.0

[hadoop@node02 azkaban-exec-server-4.0.0]$ vim tmp.sh

[hadoop@node03 azkaban-exec-server-4.0.0]$ cd

[hadoop@node03 ~]$ cd /export/servers/azkaban-exec-server-4.0.0

[hadoop@node03 azkaban-exec-server-4.0.0]$ vim tmp.sh

脚本内容如下

1 2	#!/bin/bash echo 'this is echoed by tmp sh script'

方法二：重试flow时，指定执行的executor（此处暂略，下文会提到用法）

手动重试有两个方案

1、方案一

2、方案二

Enable 和 Disable 下面都分别有如下参数：
Parents：该作业的上一个job
Ancestors：该作业前的所有job
Children：该作业后的一个job
Descendents：该作业后的所有job
Enable All：所有的job

可以根据实际情况选择enable方案

6. 操作HDFS

node01节点用root用户启动hadoop集群

[hadoop@node01 bin]$ su root

密码：

[root@node01 bin]#

[root@node01 bin]# cd

[root@node01 ~]# start-all.sh

编写flow文件operateHdfs.flow，内容如下

nodes:

- name: jobA

type: command

config:

command: echo "start execute"

command.1: /export/servers/hadoop-2.7.5/bin/hdfs dfs -mkdir /azkaban

command.2: /export/servers/hadoop-2.7.5/bin/hdfs dfs -put /export/servers/hadoop-2.7.5/NOTICE.txt /azkaban

生成zip项目文件、web ui上传zip、执行flow
查看HDFS结果

7. MR任务

记得启动hadoop的historyserver，否则执行mr项目时，job的日志会报如下类似错误日志

22-03-2021 23:17:23 CST jobMR INFO - 21/03/22 23:17:23 INFO impl.YarnClientImpl: Submitted application application_1616423563192_0001

22-03-2021 23:17:39 CST jobMR INFO - 21/03/22 23:17:39 INFO mapred.ClientServiceDelegate: Application state is completed. FinalApplicationStatus=SUCCEEDED. Redirecting to job history server

22-03-2021 23:17:41 CST jobMR INFO - 21/03/22 23:17:41 INFO ipc.Client: Retrying connect to server: node01/192.168.77.30:10020. Already tried 0 time(s); retry policy is RetryUpToMaximumCountWithFixedSleep(maxRetries=10, sleepTime=1000 MILLISECONDS)

22-03-2021 23:17:42 CST jobMR INFO - 21/03/22 23:17:42 INFO ipc.Client: Retrying connect to server: node01/192.168.77.30:10020. Already tried 1 time(s); retry policy is RetryUpToMaximumCountWithFixedSleep(maxRetries=10, sleepTime=1000 MILLISECONDS)

22-03-2021 23:17:44 CST jobMR INFO - 21/03/22 23:17:44 INFO ipc.Client: Retrying connect to server: node01/192.168.77.30:10020. Already tried 2 time(s); retry policy is RetryUpToMaximumCountWithFixedSleep(maxRetries=10, sleepTime=1000 MILLISECONDS)