首页 > 代码库 > Azkaban简介

Azkaban简介

  1. 一、Azkaban简介  

  2. Azkaban作为开源的调度系统,在大数据中有广泛地使用。它主要有三部分组成:Azkaban Webserver、Azkaban Executor、 DB。

  3. 技术分享

  4.               图1 Azkaban架构

  5. 图1所示的是Azkaban的基本架构:Webserver主要负责权限验证、项目管理、作业流下发等工作;Executor主要负责作业流/作业的具体执行以及搜集执行日志等工作;MySQL用于存储作业/作业流的执行状态信息。图中所示的是单executor场景,但是实际应用中大部分的项目使用的都是多executor场景。下面主要介绍多executor场景下的azkaban调度过程。

  6. 二、作业流执行过程

  7. 技术分享

  8.                      图2 作业流执行过程

  9. 图2展示的就是Azkaban作业流的执行过程:

  10. 1. 首先Webserver根据内存中缓存的各Executor的资源状态(Webserver有一个线程会遍历各个active executor,去发送http请求获取其资源状态信息缓存到内存中),按照选择策略(包括executor资源状态、最近执行流个数等)选择一个executor下发作业流;

  11. 2. 然后executor判断是否设置作业粒度分配,如果未设置作业粒度分配,则在当前executor执行所有作业;

  12. 3. 如果设置了作业粒度分配,则当前节点会成为作业分配的决策者,即分配节点;

  13. 4. 分配节点从zookeeper获取各个executor的资源状态信息,然后根据策略选择一个executor分配作业;

  14. 5. 被分配到作业的executor即成为执行节点,执行作业,然后更新数据库。

  15. 三、从源码看作业流执行过程

  16. 首先是Webserver端:

  17. 1. ExecutorServlet类根据请求的ajax参数判断,如果ajax=executeFlow,就去调ajaxAttemptExecuteFlow(req, resp, ret, session.getUser())方法

  18. 2. ajaxAttemptExecuteFlow方法里,首先调getProjectAjaxByPermission方法判断用户是否有执行权限,如果验证权限通过,且Project和Flow都存在,就调ajaxExecuteFlow方法

  19. 3. ajaxExecuteFlow方法的主要作用就是构造ExecutableFlow对象,设定执行参数(通知机制,并发,失败策略),然后去调executorManager.submitExecutableFlow方法

  20. 4. executorManager.submitExecutableFlow方法:判断执行策略(流水线、忽略、并发);如果是多执行节点模式,则将作业流提交到执行队列queue;如果是单执行节点模式,选择唯一执行节点下发作业流。

  21. 5. ExecutorManager.submitExecutableFlow()方法是Webserver端下发作业流的主要实现逻辑,下面重点细述其内容:

  22.     5.1 从exflow实例获取作业流的flowId(就是作业流的名字),打日志(“开始提交流XXX by 某某某了”)。

  23.     5.2 判断queuedFlows是否满,如果满了打日志(“提交失败,Azkaban过饱和啦”),return;如果未满,继续往下执行代码

  24.     5.3 获取该作业流所有正在跑的实例的id, List<Integer> running
        5.4 获取执行设置options
        5.5 从执行设置options里获取流的执行参数(是否enable,是则将参数生效)
        5.6 判断running是否为空,如果为空,即没有并发的实例在跑
        5.7 如果running不为空,获取并发设置getConcurrentOption()
             5.7.1 流水线(pipeline):设置pipelineExcutionId为running中最后提交的实例id
             5.7.2 忽略(skip):抛异常,“流已经在执行了,忽略本次执行”
             5.7.3 并发(ignore):仅修改日志
        5.8 根据白名单设置是否memoryCheck
        5.9 executorLoader.uploadExecutableFlow(exflow) 写数据库表execution_flows,状态为preparing

  25.     5.10 构造具体的执行实例ExecutionReference
        5.11 判断是否多执行节点模式,如果不是,将该执行流的状态标记为active,即写数据库表active_executing_flows,将流dispatch到唯一执行节点执行。
        5.12 如果是多执行节点模式,则将该执行流的状态标记为active,然后将流放入执行队列queuedFlows。

  26. 6. 如果是多执行节点模式,ExecutorManager类在构造函数里会调setupMultiExecutorMode()方法,该方法会建一个线程通过processQueuedFlows方法去持续地消费队列里的首个作业流。processQueuedFlows方法的主要内容就是按照一定规则去refreshExecutors刷新执行节点的资源信息,以及selectExecutorAndDispatchFlow从activeExecutors中根据策略选择一个executor下发作业流。refreshExecutors()方法实际上是通过遍历每个active executor,去发请求获取状态信息,而不是通过zookeeper。

  27. 至此,Webserver端的工作已经完毕。

  28. 然后是Executor端:

  29. 1. 执行流到达Executor端,此时在数据库中的状态已经是preparing

  30. 2. ExecutorServlet类根据请求的action参数判断,如果action=execute,就去调handleAjaxExecute(req, respMap, execid)方法

  31. 3. handleAjaxExecute方法里执行flowRunnerManager.submitFlow(execId),去调FlowRunnerManager的submitFlow(execId)方法来提交执行流。

  32. 4. FlowRunnerManager的两个重要的数据结构:

  33.     4.1 Map<Future<?>, Integer> submittedFlows = new ConcurrentHashMap<Future<?>, Integer>();

  34.     4.2 Map<Integer, FlowRunner> runningFlows = new ConcurrentHashMap<Integer, FlowRunner>();

  35.     submittedFlows用于跟踪当前executor所有处于preparing状态的流的执行;runningFlows用于存数当前executor所有正在执行的流的信息,当需要执行cancling()或killing()的时候就可以找到这些流。

  36. 5. FlowRunnerManager.submitFlow(execId)方法是Executor端执行作业流的主要实现逻辑,下面重点细述其内容:

  37.     5.1 先判断runningFlows是否包含该execId对应的实例,如果已经包含,抛异常

  38.     5.2 从executorLoader去获取execId对应的执行实例(ExecutableFlow)flow

  39.     5.3 执行setupFlow(flow),配置flow:创建项目和执行的目录等
        5.4 获取执行设置ExecutionOptions
        5.5 判断pipelineExecId是否为null。如果不为null,就判断pipelineExecId对应的flowRunner在不在runningFlows中。如果在runningFlows中,起一个LocalFlowWatcher去监控在flow中各个job的执行状态;

  40.     5.6 如果不在runningFlows中,起一个RemoteFlowWatcher去监控,即每隔一定时间(默认为60秒)通过读取数据库的记录来监控流中各个job的状态

  41.     5.7 判断执行参数里是否包含flow.num.job.threads,如果存在且小于默认值10,则修改该值。这个值代表该流可以同时执行的job线程数。
        5.8 构造一个新的FlowRunner实例runner
        5.9 configureFlowLevelMetrics(runner)配置runner 
        5.10 再次判断runningFlows是否包含该次execId对应的执行实例,如果包含,抛异常
        5.11 将runner加入到runningFlows的map
        5.12 提交到TrackingThreadPool(工作线程池)
        5.13 加入到submittedFlows的map

  42. 6. 自此,我们就有了FlowRunner实例,下面我们看FlowRunner中都干了些什么事。

  43. FlowRunner其实就是一个线程,它的run()方法的内容如下:

  44.     6.1 Executors.newFixedThreadPool(numJobThreads) 创建flow内部job线程池flow
        6.2 setupFlowExecution()
        6.3 updateFlowReference()
        6.4 updateFlow() 更新flow的状态信息,写数据库表execution_flows
        6.5 loadAllProperties()载入job参数和共享的参数
        6.6 判断输入参数是否包含job.dispatch(作业粒度分配),如果包含且为true,起一个新的线程jobEventUpdaterThread,用于跟踪该作业流下各个作业的执行状态。
        6.7 执行runFlow()
        6.8 runFlow()方法:根据DAG图的算法依次执行job。从流的开始节点,递归调用runReadyjob()来执行作业,然后updateFlow();如果流还没结束,根据重试设置,决定是否重跑失败的作业。
        6.9 在runReadyjob()里会调runExecutableNode(node)方法,runExecutableNode方法再判断job.dispatch参数,如果为false,则通过LocalJobRunner本地执行;如果为true,则再通过JobRunnerManager提交作业。
        6.10 JobRunnerManager通过submitExecutableNode方法构建RemoteJobRunner,RemoteJobRunner会根据各执行节点(包含本节点)的资源状态去选择一个节点执行作业。

  45.  最后,整个过程可以总结成一个图,如下图所示:

技术分享

                                                                       图3 从源码看作业流执行过程


Azkaban简介