全流程调度

#全流程调度| 来源: 网络整理| 查看: 265

文章目录概述为什么需要工作流调度系统常见工作流调度系统Azkaban与Oozie对比架构Azkaban作业流执行过程Azkaban架构的三种运行模式核心调度概述整体概述选择Executor Server获取Executor Server的运行统计信息调度WorkFlow到Executor Server上执行Executor Server执行WorkFlow Azkaban入门集群模式安装上传tar包配置MySQL配置Executor Server配置Web Server Work Flow案例实操HelloWorld案例作业依赖案例自动失败重试案例手动失败重试案例 Azkaban进阶JavaProcess作业类型案例条件工作流案例运行时参数案例预定义宏案例定时执行案例邮件报警案例电话报警案例第三方告警平台集成测试 Azkaban多Executor模式注意事项项目调度数据准备编写Azkaban工作流程配置文件

概述

Azkaban是由Linkedin公司推出的一个批量工作流任务调度器，用于在一个工作流内以一个特定的顺序运行一组工作和流程。Azkaban使用job配置文件建立任务之间的依赖关系，并提供一个易于使用的web用户界面维护和跟踪你的工作流。

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

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

总体来说，Ooize相比Azkaban是一个重量级的任务调度系统，功能全面，但配置使用也更复杂。如果可以不在意某些功能的缺失，轻量级调度器Azkaban是很不错的候选对象。

架构

（1）AzkabanWebServer

**AzkabanWebServer是整个Azkaban工作流系统的主要管理者，它负责project管理、用户登录认证、定时执行工作流、跟踪工作流执行进度等一系列任务。**同时，它还提供Web服务操作的接口，利用该接口，用户可以使用curl或其他ajax的方式，来执行azkaban的相关操作。

操作包括：用户登录、创建project、上传workflow、执行workflow、查询workflow的执行进度、杀掉workflow等一系列操作，且这些操作的返回结果均是json的格式。并且Azkaban使用方便，Azkaban使用以.job为后缀名的键值属性文件来定义工作流中的各个任务，以及使用dependencies属性来定义作业间的依赖关系链。这些作业文件和关联的代码最终以*.zip的方式通过Azkaban UI上传到Web服务器上。

（2）AzkabanExecutorServer

以前版本的Azkaban在单个服务中具有AzkabanWebServer和AzkabanExecutorServer功能，目前Azkaban已将AzkabanExecutorServer分离成独立的服务器，拆分AzkabanExecutorServer的原因有如下几点：

某个任务流失败后，可以更方便的将其重新执行便于Azkaban升级

AzkabanExecutorServer主要负责具体的工作流的提交、执行，可以启动多个执行服务器，它们通过mysql数据库来协调任务的执行以及实现高可用性。

（3）关系型数据库（MySQL）

Azkaban使用数据库存储大部分状态，AzkabanWebServer和AzkabanExecutorServer都需要访问数据库。

AzkabanWebServer使用数据库的原因如下：

项目管理：项目、项目权限以及上传的文件。执行流状态：跟踪执行流程以及执行程序正在运行的流程。以前的流程/作业：通过以前的作业和流程执行以及访问其日志文件进行搜索。计划程序：保留计划作业的状态。SLA：保持所有的SLA规则

AzkabanExecutorServer使用数据库的原因如下：

访问项目：从数据库检索项目文件。执行流程/作业：检索和更新正在执行的作业流的数据日志：将作业和工作流的输出日志存储到数据库中。交互依赖关系：如果一个工作流在不同的执行器上运行，它将从数据库中获取状态 Azkaban作业流执行过程

Webserver根据内存中缓存的各Executor的资源状态（Webserver有一个线程会遍历各个active executor，去发送http请求获取其资源状态信息缓存到内存中），按照选择策略（包括executor资源状态、最近执行流个数等）选择一个executor下发作业流；executor判断是否设置作业粒度分配：

如果未设置作业粒度分配，则在当前executor执行所有作业；如果设置了作业粒度分配，则当前节点会成为作业分配的决策者，即分配节点；分配节点从zookeeper获取各个executor的资源状态信息，然后根据策略选择一个executor分配作业；被分配到作业的executor即成为执行节点，执行作业，然后更新数据库。 Azkaban架构的三种运行模式

在版本3.0中，Azkaban提供了以下三种模式：

solo server mode：最简单的模式，数据库内置的H2数据库，AzkabanWebServer和AzkabanExecutorServer都在一个进程中运行，任务量不大项目可以采用此模式。

two server mode：数据库为MySQL，管理服务器和执行服务器在不同进程，这种模式下，AzkabanWebServer和AzkabanExecutorServer互不影响。

multiple executor mode：该模式下，AzkabanWebServer和AzkabanExecutorServer运行在不同主机上，且AzkabanExecutorServer可以有多个。大数据平台要求其具有高可用性，所以目前我们采用的是multiple executor mode方式，分别在不同的主机上部署多个AzkabanExecutorServer以应对高并发定时任务执行的情况，从而减轻单个服务器的压力。下面是集群架构图：

核心调度概述整体概述

Azkaban WebServer需要根据Executor Server的运行状态信息，选择一个合适的Executor Server来运行WorkFlow，然后会将提交到队列中的WorkFlow调度到选定的Executor Server上运行。我们整理了与核心调度相关的各个组件，主要包括Azkaban WebServer端和Azkaban ExecutorServer端，他们之间的关系如下图所示：

其实，从调度层面来看，Azkaban WebServer与Executor Server之间的交互方式非常简单，是通过REST API的方式来进行交互，基本的模式是，Azkaban WebServer根据调度的需要，主动调用Executor Server暴露的REST API来获取相应的资源信息，比如Executor Server的状态信息、分配WorkFlow到指定Executor Server上运行，等等。

我们可以在QueueProcessorThread.selectExecutorAndDispatchFlow()方法中看到，选择Executor Server并进行调度的实现，代码片段如下所示：

final Executor selectedExecutor = selectExecutor(exflow, availableExecutors); if (selectedExecutor != null) { try { dispatch(reference, exflow, selectedExecutor); ExecutorManager.this.commonMetrics.markDispatchSuccess(); } catch (final ExecutorManagerException e) { ExecutorManager.this.commonMetrics.markDispatchFail(); logger.warn(String.format( "Executor %s responded with exception for exec: %d", selectedExecutor, exflow.getExecutionId()), e); handleDispatchExceptionCase(reference, exflow, selectedExecutor, availableExecutors); } }

QueueProcessorThread是运行在Azkaban WebServer端的一个线程，它在ExecutorManager中定义，是内部调度中最核心的线程。

selectExecutor()方法处理如何选择一个合适的Executor Server，然后通过dispatch()方法将需要运行的WorkFlow调度到该Executor Server上运行。

选择Executor Server

Azkaban WebServer选择Executor，调用selectExecutor()方法，实现如下所示：

private Executor selectExecutor(final ExecutableFlow exflow, final Set availableExecutors) { Executor choosenExecutor = getUserSpecifiedExecutor(exflow.getExecutionOptions(), exflow.getExecutionId()); // If no executor was specified by admin if (choosenExecutor == null) { logger.info("Using dispatcher for execution id :" + exflow.getExecutionId()); final ExecutorSelector selector = new ExecutorSelector(ExecutorManager.this.filterList, ExecutorManager.this.comparatorWeightsMap); choosenExecutor = selector.getBest(availableExecutors, exflow); } return choosenExecutor; } 首先，查看当前exflow的配置中，是否要求将该exflow调度到指定的Executor Server上运行，如果是的话，则会返回该指定的Executor Server的信息，后续直接调度到该Executor Server上；否则会按照一定的计算规则去选出一个Executor Server。在创建ExecutorSelector时，传入参数值ExecutorManager.this.filterList，该filterList是从azkanban.properties文件中读取azkaban.executorselector.filters的配置值，并创建了一个ExecutorFilter对象，而该对象中包含了一组FactorFilter。使用ExecutorSelector来选出一个Executor Server，具体选择的逻辑，我们可以查看ExecutorSelector.getBest()方法。

首先通过定义的CandidateFilter（它是一个抽象类，具体实现类为ExecutorFilter）进行预筛选：

for (final K candidateInfo : candidateList) { if (this.filter.filterTarget(candidateInfo, dispatchingObject)) { filteredList.add(candidateInfo); } }

上面的filter就是FactorFilter类的实例，Azkaban内部定义了如下3种：

private static final String STATICREMAININGFLOWSIZE_FILTER_NAME = "StaticRemainingFlowSize"; private static final String MINIMUMFREEMEMORY_FILTER_NAME = "MinimumFreeMemory"; private static final String CPUSTATUS_FILTER_NAME = "CpuStatus";

目前3.40.0版本不支持自定义，只能使用内建实现的，如果需要增加新的FactorFilter，可以在此基础上做一个简单改造，配置使用自己定义的FactorFilter实现。FactorFilter是一个泛型类：FactorFilter，根据上面定义的3种指标对Executor Server进行一个预过滤，满足要求的会进行后面的比较，加入到调度WorkFlow执行的Executor Server的候选集中。

然后，通过如下方式进行比较排序，选择合适的Executor Server：

// final work - find the best candidate from the filtered list. final K executor = Collections.max(filteredList, this.comparator); logger.debug(String.format("candidate selected %s", null == executor ? "(null)" : executor.toString())); return executor;

这里关键的就是this.comparator，它有一个实现类ExecutorComparator，该类中给出了需要对两个Executor Server的哪些指标进行综合比较，亦即一组比较器的定义，可以看到目前考虑了4种比较器：

private static final String NUMOFASSIGNEDFLOW_COMPARATOR_NAME = "NumberOfAssignedFlowComparator"; private static final String MEMORY_COMPARATOR_NAME = "Memory"; private static final String LSTDISPATCHED_COMPARATOR_NAME = "LastDispatched"; private static final String CPUUSAGE_COMPARATOR_NAME = "CpuUsage";

通过上面代码可以看出，在选择调度一个WorkFlow到Azkaban集群中的某个Executor Server时，需要比较Executor Server的如下4个指标：

能够运行WorkFlow的剩余容量，数值越大越优先剩余内存用量，数值越大越优先最近分配Flow的时间，数值越大越优先CPU使用用量，数值越小越优先

基于上面4个指标，创建了4个比较器，使用FactorComparator来表示，对需要比较的一组Executor Server，使用这4个比较器进行比较，通过加权后得到一个得分值，根据该得分值选定Executor Server，核心逻辑如下所示：

final Collection comparatorList = this.factorComparatorList.values(); for (final FactorComparator comparator : comparatorList) { final int result = comparator.compare(object1, object2); result1 = result1 + (result > 0 ? comparator.getWeight() : 0); result2 = result2 + (result activeExecutorsRefreshWindow || currentContinuousFlowProcessed >= maxContinuousFlowProcessed) { // Refresh executorInfo for all activeExecutors refreshExecutors(); lastExecutorRefreshTime = currentTime; currentContinuousFlowProcessed = 0; }

上面**refreshExecutors()**方法遍历内存中维护的所有的Executor Server，调用每个Executor Server的/serverStatistics接口，拉取Executor Server的运行状态信息。

另外，Azkaban WebServer还需要能够获取到各个Executor Server上运行的WorkFlow的状态信息，可以在ExecutorManager.ExecutingManagerUpdaterThread中看到实现，代码片段如下所示：

results = ExecutorManager.this.apiGateway.callWithExecutionId(executor.getHost(), executor.getPort(), ConnectorParams.UPDATE_ACTION, null, null, executionIds, updateTimes);

上面调用Executor Server的/executor?action=update接口来拉取WorkFlow的状态信息，然后更新内存中维护的状态信息数据结构。其中，有些WorkFlow可能已经运行完成，需要释放资源；有些WorkFlow状态发生变更，也需要更新Azkaban WebServer端内存中维护的数据结构。

有三种API：

public static final String DEFAULT_CLUSTER_NAME = "azkaban"; public final static String DEFAULT_EXECUTION_RESOURCE = "executor"; public final static String CONTAINERIZED_EXECUTION_RESOURCE = "container"; 调度WorkFlow到Executor Server上执行

上面已经选定Executor Server，结合前面代码，是通过调用ExecutorManager.dispatch()方法来实现，调度WorkFlow到该选定的Executor Server上运行，代码片段如下所示：

try { this.apiGateway.callWithExecutable(exflow, choosenExecutor, ConnectorParams.EXECUTE_ACTION); } catch (final ExecutorManagerException ex) { logger.error("Rolling back executor assignment for execution id:" + exflow.getExecutionId(), ex); this.executorLoader.unassignExecutor(exflow.getExecutionId()); throw new ExecutorManagerException(ex); }

通过跟踪查看apiGateway.callWithExecutable()实现，可以看到，最终是调用了Executor Server端的一个REST API接口：/executor，然后带上相关的请求参数，如action=execute、execId等。

Executor Server执行WorkFlow

很显然，Azkaban WebServer调度WorkFlow后，Executor Server在ExecutorServlet中接收到对应的请求，核心方法如下所示：

private void handleAjaxExecute(final HttpServletRequest req, final Map respMap, final int execId) throws ServletException { try { this.flowRunnerManager.submitFlow(execId); } catch (final ExecutorManagerException e) { e.printStackTrace(); logger.error(e.getMessage(), e); respMap.put(RESPONSE_ERROR, e.getMessage()); } }

在收到Azkaban WebServer的调度请求后，Executor Server使用内部的FlowRunnerManager来提交WorkFlow执行。在这个过程中，首先使用ExecutorLoader从数据库中读取WorkFlow对应的信息；然后使用FlowPreparer进行初始化，创建对应的数据目录等；最后创建FlowRunner来执行WorkFlow，并跟踪其执行状态。

Azkaban入门集群模式安装上传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路径

[atguigu@hadoop102 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包解压到这个目录下

[atguigu@hadoop102 software]$ 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目录下

[atguigu@hadoop102 software]$ tar -zxvf azkaban-db-3.84.4.tar.gz -C /opt/module/azkaban/ [atguigu@hadoop102 software]$ tar -zxvf azkaban-exec-server-3.84.4.tar.gz -C /opt/module/azkaban/ [atguigu@hadoop102 software]$ tar -zxvf azkaban-web-server-3.84.4.tar.gz -C /opt/module/azkaban/

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

[atguigu@hadoop102 azkaban]$ mv azkaban-exec-server-3.84.4/ azkaban-exec [atguigu@hadoop102 azkaban]$ mv azkaban-web-server-3.84.4/ azkaban-web 配置MySQL

（1）正常安装MySQL

（2）启动MySQL

[atguigu@hadoop102 azkaban]$ 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阻塞

[atguigu@hadoop102 software]$ sudo vim /etc/my.cnf

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

[mysqld] max_allowed_packet=1024M

（8）重启MySQL

[atguigu@hadoop102 software]$ sudo systemctl restart mysqld 配置Executor Server

Azkaban Executor Server处理工作流和作业的实际执行。

（1）编辑azkaban.properties

[atguigu@hadoop102 azkaban]$ 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到所有节点

[atguigu@hadoop102 azkaban]$ xsync /opt/module/azkaban/azkaban-exec

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

[atguigu@hadoop102 azkaban-exec]$ bin/start-exec.sh [atguigu@hadoop103 azkaban-exec]$ bin/start-exec.sh [atguigu@hadoop104 azkaban-exec]$ bin/start-exec.sh

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

4）下面激活executor，需要

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

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

{"status":"success"} 配置Web Server

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

（1）编辑azkaban.properties

[atguigu@hadoop102 azkaban]$ 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文件，添加atguigu用户

[atguigu@hadoop102 azkaban-web]$ vim /opt/module/azkaban/azkaban-web/conf/azkaban-users.xml

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

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

（4）访问http://hadoop102:8081，并用atguigu用户登陆

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" Name：job名称Type：job类型。command表示你要执行作业的方式为命令Config：job配置

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

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

（5）给项目名称命名和添加项目描述

（6）first.zip文件上传

（7）选择上传的文件

（8）执行任务流

（9）在日志中，查看运行结果

作业依赖案例

需求：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"

dependsOn：作业依赖，后面案例中演示

（2）将修改后的basic.flow和azkaban.project压缩成second.zip文件

（3）重复前面HelloWorld后续步骤

自动失败重试案例

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

具体步骤：

（1）编译配置流

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

参数说明：

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

（2）将修改后的basic.flow和azkaban.project压缩成four.zip文件

（3）重复前面HelloWorld后续步骤。

（4）执行并观察到一次失败+三次重试

（5）也可以点击上图中的Log，在任务日志中看到，总共执行了4次。

（6）也可以在Flow全局配置中添加任务失败重试配置，此时重试配置会应用到所有Job。

案例如下：

config: retries: 3 retry.backoff: 10000 nodes: - name: JobA type: command config: command: sh /not_exists.sh 手动失败重试案例

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

具体步骤：

（1）编译配置流

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."

（2）将修改后的basic.flow和azkaban.project压缩成five.zip文件

（3）重复前面HelloWorld后续步骤

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

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

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

Azkaban进阶 JavaProcess作业类型案例

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

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

案例：

（1）新建一个azkaban的maven工程

（2）创建包名：com.atguigu

（3）创建AzTest类

package com.atguigu; 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.atguigu.AzTest

（6）将Jar包、flow文件和project文件打包成javatest.zip

（7）创建项目=》上传javatest.zip =》执行作业=》观察结果

条件工作流案例

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

运行时参数案例

（1）基本原理

父Job将参数写入JOB_OUTPUT_PROP_FILE环境变量所指向的文件子Job使用 ${jobName:param}来获取父Job输出的参数并定义执行条件

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

== 等于!= 不等于> 大于>= 大于等于< 小于

【本文地址】

全流程调度

全流程调度

今日新闻

推荐新闻