基于异构Flink 集群的节点优先级调度策略

汪文豪，史雪荣

（1.南京工业大学 计算机科学与技术学院，南京 211816；2.盐城师范学院 数学与统计学院，江苏 盐城 224002）

0 概述

近年来，云计算、大数据、物联网、人工智能等新一代信息技术快速发展，信息产业的数据量急剧增长。截止到2020 年底，我国的数据总量预计到达8 060 EB［1］。随着信息数据量剧增，数据处理技术也在发生着巨大变化，涌现出Apache Hadoop、Apache Spark、Apache Storm、Heron、Flink 等［2］一大批数据计算系统。在数据处理的硬件方面，各种机构和设备的更新换代和新兴技术的引入，使得各个数据计算系统在实际生产环境中变得越来越异构［3-4］。这种异构集群特征直观表现在其节点的CPU、内存等各方面存在差异致使集群运行时的处理能力不尽相同，从而使得集群在调度节点执行任务时，集群效率明显下降［5］。

目前，国内外众多学者对资源弹性管理问题和分布式框架下的任务及节点调度问题展开研究。在资源弹性管理方面：文献［6］提出使用排队理论进行建模，通过部署主动弹性控制器和反应弹性控制器相结合的模型来估计集群节点的未来负载；文献［7］认为云计算存在资源倾斜的问题，提出动态实时云框架，使集群可以自动适应不同的工作负载，并根据需求重新分配资源；文献［8］基于云计算通用工作负载预测器，设计一种可以为不同工作负载提供高精度预测的长期记忆模型，解决了通用预测器精度不够的问题，进一步优化了集群资源弹性管理。在批处理框架方面：文献［9］认为Hadoop 框架在默认调度时未考虑Map 与Reduce 之间存在差异性，据此提出一种基于Hadoop 框架的截止时间约束的扩展MapReduce 任务调度算法；文献［10］认为在Hadoop默认调度方式下缺乏对整体集群异构性的考虑，提出基于资源感知与资源异构的云计算环境任务调度算法；文献［11］基于Hadoop 框架的节点计算能力，设计能够按计算能力分配数据块的数据局部性调度器；文献［12］认为在异构Hadoop 集群中会出现节点随任务动态分配而产生性能差异的问题，提出基于动态工作负载调整的任务调度策略；文献［13］在YARN 资源调度器的基础上，结合闭环反馈控制方法，使Hadoop 集群在运行时可以动态对MapReduce作业数进行控制，避免出现主观性的问题；文献［14］基于异构Spark 集群，提出通过监测节点资源的自适应动态节点任务调度策略，但是该策略依赖于人工设定的阈值，无法体现客观性；文献［15］在文献［14］的基础上，提出基于异构节点优先级的Spark 动态自适应调度算法，该算法能够根据实时节点优先级完成调度，但是忽略了任务种类和集群整体工作环境，没有彻底解决主观性问题。在流处理框架方面：文献［16］提出在流处理框架Storm on YARN 上构建一种双层调度模型，通过对流数据处理的实时监测，做出合理的资源分配预测，解决框架需要人工干预调整的问题；文献［17］针对流处理框架的数据拓扑中任务间通信开销较大的问题，提出Flink 框架下的拓扑关键路径模型，该模型能够确保关键路径上节点负载差异较小的同时最小化任务的节点间通信开销。

但是现阶段国内外的相关研究主要存在两方面的问题。一方面，大部分的分布式调度研究主要关注于批处理，将Hadoop 框架与Spark 框架作为主要的实验环境，而对于流处理，尤其是以Flink 框架作为实验环境的研究相对较少。另一方面，在实际作业环境中，很难避免整体集群出现异构的情况，针对结合实际作业环境的异构Flink 环境的研究也相对较少。本文通过研究分布式节点的实时性能和集群的工作环境预测机制，充分考虑现实生产环境的异构分布问题，从而对Flink 底层默认调度策略进行优化，以提高Flink 系统的工作效率。

1 相关技术

Apache Flink 是一个面向数据流的有状态计算框架，核心是一个流数据的处理引擎。在计算过程中，Flink 将所有任务当作流来处理，批处理任务被看作具备有限边界的特殊数据流。相对于目前的大部分流处理框架，Flink 框架在数据处理方面有低延迟、高吞吐的优势，在集群功能方面则提供了消息传递、状态管理、容错恢复等一系列服务［18-19］。

Flink 集群主要由两个重要进程JobManager 和TaskManager 组成，如图1 所示，其中虚线表示任务的流转。JobManager 又称为Master，主要协调整体架构的数据流图的运算和执行，其中的调度器模块和Checkpoint 协调器模块还负责调度任务、协调检查和故障恢复。TaskManager又称为Worker，主要执行由JobManager分配的任务。同时，每一个TaskManager都具有缓存数据和节点间通信的功能。在Flink 中，一个TaskManager 即为一个JVM 进程，JVM 进程允许并行地执行子任务，并能够指定若干slot。每一个slot 可以执行若干子任务，即运行若干线程［18］。

图1 Flink 集群架构Fig.1 Flink cluster architecture

Flink 在调度任务分配slot 时，遵循2 个原则：1）同一Job 中的同一分组的不同task 可以共享同一slot；2）任务调度按照拓扑顺序从Source 调度到Sink。以图2（a）的Flink 拓扑模型为例，各顶点ν表示Flink 中的Operator 算子，νa表示Source 算子组件，νb、νc和νd表示并行度为3 的Transformation 算子组件，νe和νf表示并行度为2 的Sink 算子组件。首先由Source 组件将读取的数据发送至Transformation 组件；然后Transformation 组件负责处理上游发送的数据；最后由Sink 组件接收上游处理结果并持久化至数据库［20］。以有2 个TaskManager 的Flink 分布式集群为例，每一个TaskManager 配置有2 个slot。Flink在默认调度下对于该任务拓扑的slot 分配模型如图2（b）所示。首先调度默认从拓扑的Source 任务开始，将νa随机调度给任一slot，如图2（b）中νa调度至slot11 所示；然后将νb、νc和νd调度至任意的slot，但νb、νc和νd同属于一种Operator 算子，不能共享同一slot，因此将νb、νc和νd分别调度至slot11、slot12 和slot21；最后将νe和νf也分别调度至slot11 和slot12。

图2 任务拓扑分配模型Fig.2 Task topology assignment model

2 基于节点优先级的Flink 动态调度策略

2.1 节点优先级定义及调整方法描述

2.1.1 相关定义

定义1（节点优先级）Flink 集群的节点优先级集合为P=｛P1，P2，…｝。Pix表示第i个节点的偏x相关性能优先级指数，当x=c时表示节点偏CPU 相关性能优先级指数，当x=m时表示节点偏内存相关性能优先级指数。

定义2（节点性能指数计算权值）在节点性能指数计算时，引入的各动静态性能因素权值表示为向量与向量，当x=c时当前权值侧重计算CPU相关优先性能指数，当x=m时当前权值侧重计算内存相关优先性能指数。对于任意的，有；对于任意的。

定义3（节点资源因素指数［15］）动态性能指数D=｛D1，D2，…｝表示节点在执行任务时的实时性能变化。对于动态性能指数有3 个资源约束，分别是CPU 剩余率（CSR）、内存剩余率（MSR）和磁盘剩余率（DSR）。静态性能指数S=｛S1，S2，…｝表示节点的不对称性能因素。对于静态性能指数有4 个资源约束，分别为CPU 速度（CS）、CPU 核数（CQ）、内存容量（MC）和磁盘容量（DC）。

对于每个动态指数Di∈D，对应资源约束向量di=(CCSRi，MMSRi，DDSRi)；对于每个静态指数Si∈S，对应资源约束向量si=(CCSi，CCQi，MMCi，DDCi)。引入定义2中权值的计算后，得到Dix表示第i个节点的动态偏x性能指数，Six表示第i个节点的静态偏x性能指数。

2.1.2 调整方法描述

本文所设计的节点优先级调整方法是即时反馈方法，节点优先级计算公式如下：

其中：α、β为节点整体动静态因素指数权值，α+β=1；和表示节点动静态偏x性能指数。和计算公式如下：

以4 个节点组成的Flink 异构集群为例，给出节点优先级调整架构如图3 所示。

图3 节点优先级调整架构Fig.3 Node priority adjustment architecture

监控器运行在集群Master 节点中，负责周期性收集节点的资源使用情况和任务队列长度。获取的资源信息将传入到控制器中，由控制器完成具体节点优先级计算。监控器采用Ganglia［21］进行实现，Ganglia 是一个可扩展分布式集群资源监控系统，能够实现对集群信息的监控。

控制器负责根据监控器监测到的信息计算出各节点的优先级。本文节点优先级调整方法引入作业环境预测机制，在一个周期内判定当前集群作业环境的参数x由式（4）确定。如果集群节点的I/O 操作平均时间T（I/O）超过了整体操作时间T（all）的65%，则判定当前集群作业环境为I/O 密集型，否则判定为CPU 密集型。

当监控器判定为I/O 密集型环境时，控制器将节点动静态因素指数计算的权值重置为和，节点的偏内存性能优先级指数的计算公式如下：

当监控器判定为CPU 密集型环境时，控制器将节点动静态因素指数计算的权值重置为和，节点的偏CPU 性能优先级指数的计算公式如下：

2.2 节点优先级调整算法

算法1节点优先级调整算法EP-NPAA

2.3 基于节点优先级的Flink 节点动态自适应调度策略

在实际运行环境中，异构分布式集群的节点往往会出现资源不平衡和负载不均衡的情况。为保证任务的高效完成，需要准确衡量各节点的性能以及整体集群的负载程度，选择合适的节点分配任务［22］。本文提出一种基于节点优先级的Flink 节点动态自适应调度算法。

2.3.1 基于节点优先级的调度方法描述

基于节点优先级的Flink 节点动态自适应调度方法建立在集群异构的情况下，该方法的主要构成如下：1）Master 节点监控器定期监控每个Worker 节点自身资源情况及负载变化情况，并反馈控制器定期动态计算各Worker 节点的优先级指数；2）在Master 节点进行任务调度时，读取各节点优先级指数，选择指数较大的节点分配任务。Flink 节点动态自适应调度架构如图4 所示。

图4 Flink 节点动态自适应调度架构Fig.4 Dynamic adaptive Flink node scheduling architecture

2.3.2 Flink 节点动态自适应调度算法

Flink 节点动态自适应调度算法运行在Master节点上，具体步骤为：1）启动集群，Master 节点检测节点是否出现变化，根据变化状态调用EP-NPAA 算法重新计算节点的静态因素指数Si；2）Master 节点内控制器调用EP-NPAA 算法，依次计算每个Worker 节点的优先级得到P=｛P1，P2，…｝，并对节点集合NodeSets 进行排序；3）控制器从节点集合NodeSets中按优先级指数高低依次调用每个节点，检测当前调用节点是否有当前task 可用的slot，若有可用的slot，则分配任务给该节点；若无可用slot，则继续调用下一节点。

算法2Flink 节点动态自适应调度算法F-DASA

3 实验与结果分析

3.1 实验环境配置

为验证本文自适应调度策略对Flink 异构集群的节点调度性能更佳，构建Flink 异构集群数据实验平台。该平台由5 台服务器组成，其中，1 台为主服务器Master，4 台为从服务器Worker。集群节点硬件与软件配置如表1、表2 所示。

表1 集群节点硬件配置Table 1 Hardware configuration of cluster nodes

表2 集群节点软件配置Table 2 Software configuration of cluster nodes

在实验中进行节点优先级计算时，规定动静态因素权值α和β分别取值为0.65 和0.35；在内存密集型作业环境下，各动态因素权值取值为（0.3，0.6，0.1），各静态因素权值取值为（0.05，0.30，0.55，0.10）。在CPU 密集型作业环境下，各动态因素权值取值为（0.5，0.4，0.1），各静态因素权值取值为（0.10，0.50，0.35，0.05）。

3.2 实验数据设置

为有效验证F-DASA 算法对Flink 异构集群的影响，实验将本文策略与Flink 默认调度策略以及TSS-Flink 策略［17］进行大数据基准测试WorldCount和TeraSort 对比。WorldCount 是词频统计基准测试，特点是CPU 资源占用率较高、内存占用率较低，数据集采用9 个表生成模拟5 种事务处理的TPC-C 数据集［23］。TeraSort 是分布式排序基准测试，在执行过程中会大量占用内存资源，数据集为待排序的数值型数据集。

3.3 结果分析

3.3.1 运行时间对比

实验通过对比使用Flink 默认调度策略、TSSFlink 策略和本文策略后的作业运行时间，验证F-DASA 算法对Flink 异构集群的影响。在实验过程中，基准测试WorldCount 的数据集规模分别为2 GB、4 GB 和6 GB，且设置不同的作业并行度，运行时间如图5 所示。

图5 作业运行时间对比Fig.5 Comparison of job runtime

从图5 的实验结果可以看出，在并行度为8 和16 的情况下，使用TSS-Flink 策略和本文策略后，基准测试WorldCount 的运行时间都有所减少。在使用本文策略之后，比使用Flink 默认调度策略约平均减少了6%。这是因为在默认调度策略下，调度器使异构集群中资源较少的节点完成和其他节点同等的任务，拖慢了整体作业运行时间。在本文策略下，调度器使任务得到均匀分配，资源多的节点可以完成尽可能多的任务，缩短了整体运行时间。相较于TSS-Flink 策略，本文策略的运行时间更少，这是因为TSS-Flink 策略缺少对于异构环境下节点资源不均衡问题的考虑，从而导致性能有所差异。

3.3.2 系统延迟对比

实验通过对比Flink 默认调度策略与本文策略之间的延迟关系，验证F-DASA 算法对Flink 异构集群的影响。在实验过程中，在设置作业并行度为8的情况下，对比基准测试WorldCount 和TeraSort 下不同数据吞吐量的系统延迟，实验结果如图6所示。

图6 2 种策略的系统延迟对比Fig.6 Comparison of system delay of two strategies

由图6（a）可以看出，基准测试WorldCount 在使用默认调度策略时，随着吞吐量的增加，系统延迟在100 ms 至150 ms 内缓慢上升。使用本文策略后，在吞吐量为1～6 万组/s 时，系统延迟在100 ms 至150 ms 内缓慢上升，与Flink 默认调度策略的系统延迟相近；在吞吐量达到6 万组/s 以上时，系统延迟上升幅度略微增大，属于可接受范围。由图6（b）可以看出，基准测试TeraSort 在使用默认调度策略时，随着吞吐量的增加，系统延迟始终在150 ms 上下浮动。在使用本文策略后，系统延迟在吞吐量为1～6 万组/s时，在145 ms 上下浮动，略微优于Flink 默认调度策略；在吞吐量达到6 万组/s 以上时，系统延迟上升幅度略微增大，属于可接受范围。综上所述，在异构Flink 集群中使用本文策略后，依然能够保持较为稳定的延迟率，达到集群原有的响应速度。

4 结束语

针对异构Flink 集群默认策略在节点调度过程中存在部分节点负载不均衡的问题，本文提出一种基于节点优先级的Flink 节点动态自适应调度策略。该策略能够监控集群中任务与节点的各项数据，并在任务执行过程中根据实时的作业环境更新各个节点的优先级指数，为系统任务找到最佳的执行节点。实验结果表明，该策略可在保持集群低延迟的基础上，提高异构Flink 集群对于节点资源的利用率。下一步将针对节点的CPU、内存和带宽性能设置合理的阈值，确保集群不会出现满负载状态，同时设计集群任务选择算法，并将其与F-DASA 算法相结合进一步提升异构Flink 集群整体性能。