Azkaban集群内部调度原理分析

Azkaban是一个非常简单实用,而且开源的作业调度系统。在2.x版本中不支持集群模式部署,在3.x版本中支持集群模式部署,适用于作业量比较大一些的应用场景。有关Azkaban更多详细信息,如特点、功能、特性、作业定义等,可以参考官方文档,这里不再详述。

Azkaban集群架构

下面我们看一下Azkaban集群模式的架构,如下图所示:
Azkaban-Cluster-Architecture
从上图可见,Azkaban集群部署模式,主要有3个核心的组件:

  • Azkaban WebServer

Azkaban WebServer,是整个调度集群的核心,负责所有作业的管理和调度。

  • Azkaban ExecutorServer

Azkaban ExecutorServer,整个调度集群中实际运行作业的节点,该类节点可能是作为一个作业提交的客户端,比如Spark on YARN部署模式下,cluster运行模式时只作为客户端使用,client运行模式时会有部分计算逻辑;比如普通的Java程序需要处理量级较小的数据作业,这时Executor Server节点可能有较大的工作负载,占用较多节点资源(内存、CPU)。

  • DB

DB,是集群中所有节点运行共用的数据存储,包含作业信息、各种调度元数据等等。

核心调度概述

Azkaban WebServer需要根据Executor Server的运行状态信息,选择一个合适的Executor Server来运行WorkFlow,然后会将提交到队列中的WorkFlow调度到选定的Executor Server上运行。我们整理了与核心调度相关的各个组件,主要包括Azkaban WebServer端和Azkaban ExecutorServer端,他们之间的关系如下图所示:
Azkaban-Scheduling-Interaction
其实,从调度层面来看,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<Executor> 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<Executor, ExecutableFlow>,根据上面定义的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个指标:

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

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

          final Collection<FactorComparator<T>> comparatorList = this.factorComparatorList.values();
          for (final FactorComparator<T> comparator : comparatorList) {
            final int result = comparator.compare(object1, object2);
            result1 = result1 + (result > 0 ? comparator.getWeight() : 0);
            result2 = result2 + (result < 0 ? comparator.getWeight() : 0);
            logger.debug(String.format("[Factor: %s] compare result : %s (current score %s vs %s)",
                comparator.getFactorName(), result, result1, result2));
          }

上面选取了待比较的两个Executor Server都不为空的情况,分别遍历每个FactorComparator进行比较,在分别对每个Executor Server的比较结果值进行累加求和,加权得到一个分数值。从一组Executor Server中,根据最终比较的分数值,分数值最大的Executor Server为最终选定的Executor Server。

获取Executor Server的运行统计信息

在Azkaban WebServer内部,会维护集群中每个Executor Server的运行状态信息,该信息的获取是在QueueProcessorThread线程中实现的,定期去更新所维护的Executor Server的运行状态信息,如下所示:

        if (currentTime - lastExecutorRefreshTime > 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端内存中维护的数据结构。

调度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<String, Object> 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,并跟踪其执行状态。

参考内容

Creative Commons License

本文基于署名-非商业性使用-相同方式共享 4.0许可协议发布,欢迎转载、使用、重新发布,但务必保留文章署名时延军(包含链接:http://shiyanjun.cn),不得用于商业目的,基于本文修改后的作品务必以相同的许可发布。如有任何疑问,请与我联系

评论(1): “Azkaban集群内部调度原理分析

发表评论

电子邮件地址不会被公开。 必填项已用*标注

您可以使用这些HTML标签和属性: <a href="" title=""> <abbr title=""> <acronym title=""> <b> <blockquote cite=""> <cite> <code> <del datetime=""> <em> <i> <q cite=""> <strike> <strong>