基于Flink的任务调度策略

何贞贞，于 炯,, 李梓杨，国冰磊

(1.新疆大学 软件学院，新疆 乌鲁木齐 830008；2.新疆大学 信息科学与工程学院，新疆 乌鲁木齐 830046)

0 引 言

随着云计算、大数据、物联网、人工智能等信息技术的快速发展和传统产业的数字化转型，预计到2020年我国数据总量将达到8060 EB，占据全球数据总量的18%[1-4]。在这种爆炸式数据量增长情况下，其规模可以达到PB级别，产生速度可以达到GB/s级别[5]，且数据的时效性很强。对于这些连续不断的数据，目前的大多数解决方案却不是把实时流数据当作流来处理，忽略了其数据产生的连续性和及时性。为了满足这种实时性要求，流式计算[6]环境应具备较高的响应能力和较低的计算延迟，同时要求计算结果的准确性和可靠性。Apache Flink[7-13]是目前产业界应用最广泛的新兴流式计算平台之一，在美团、淘宝的实时业务中已有一定应用。

在Apache Flink中，任务是执行算子的并行化实例的基本单元，每个任务在一个进程中执行，且Flink中的任务执行图具有层次分明的拓扑结构。因此，为解决Flink计算平台拓扑中因各关键节点上任务间不同类型通信所导致的通信开销较大问题，本文提出基于一种Flink环境下的任务调度策略(task scheduling strategy in Flink，TSS-Flink)，该策略是通过动态的调整关键路径上各节点实例的任务分配，在保证关键路径节点负载差异较小的同时降低通信开销，从而降低关键路径的响应时间，提高系统性能。同时，当数据流压力发生变化后，只需要调整关键路径上部分节点任务，不会引入更多的任务调度开销。经实验验证得出，该策略对不同类型的benchmark作业都有较为明显的优化效果，在保证系统稳定性的同时使计算延迟平均降低了13.09%。

1 相关工作

在流式计算中，通常使用有向无环图(directed acyclic graph，DAG)来描述大数据流的计算过程。在拓扑中，任何情况下都会存在一条关键路径，且该关键路径的计算延迟决定了整个任务拓扑的时延。为了保证资源负载差异较小的同时提高计算的实时性，必须提出一种一方面要有效适应数据流、资源等动态变化时所带来的负载差异较大问题，另一方面也要避免因任务与任务之间不同类型的通信带来巨大的开销所造成的实时性问题。在保证各节点计算资源充分利用的前提下，最大程度降低计算延迟，提高计算实时性。

针对负载不均和任务间通信开销较大问题，已有大量的学者开展相关研究。文献[14]提出一种Storm环境下基于权重的任务调度算法，针对各个任务的CPU负载占用情况以及任务间的数据流大小，分别确定点权和边权，并利用最大化边权增益的思想，降低网络传输开销。但该算法只考虑了CPU负载对系统性能的影响，并未考虑内存资源和网络带宽资源对节点负载均衡的影响。文献[15]提出一种实时和高效的资源调度模型Re-Stream，在大数据流式计算环境下实现高能效和低延迟，并结合拓扑执行关键路径，提出对工作节点的内存电压调控节能策略，该策略主要针对关键路径和非关键路径上的内存电压节能，并未考虑到内存资源和网络带宽对系统的整体性能影响。文献[16,17]提出高效的资源调度算法和优化框架，虽然解决了流式计算框架下的任务调度问题，但无法直接移植到Flink平台。文献[18]提出一种Storm框架下资源感知的任务调度策略，通过最大化资源利用率的同时最小化网络延迟提高吞吐量。文献[19]提出用计算延迟作为评估节点间负载的指标，通过降低任务的计算延迟达到负载均衡的效果，但该策略同时也带来了较大的迁移开销，且资源评估不全面。文献[20]提出一种基于流网络的流式计算动态任务调度策略，通过定义有向无环图中每条边的容量和流量将其转化为流网络模型，计算对应的增进网络和优化路径来提升集群吞吐量从而提升性能。

现有的研究多关注于节点内的计算开销，不仅忽略了节点间的不同类型通信方式带来的传输开销，而且忽略了拓扑关键路径对集群性能的重要影响。且大多数的已有研究并不适用于Apache Flink平台。针对上述问题，本文的主要工作有：

(1)通过定义流式计算中的有向无环图(DAG)，将数据流大小作为边权将拓扑转化为AOE-网(activity on edge network)，确定拓扑中关键路径；

(2)提出负载均衡模型，主要针对关键路径上负载较高的节点，降低节点间负载差异；

(3)提出一种任务调度策略，在降低关键路径上节点过高负载的同时，最小化关键任务的节点间通信开销，即降低关键边的通信开销，实现计算资源的最大化利用。

2 关键路径检测算法

本节从Flink拓扑逻辑模型和物理模型考虑，确定任务执行拓扑关系图中关键路径，在关键路径的基础上，建立关键节点负载均衡模型和关键边最优通信开销模型，在关键路径上降低部分关键节点过高负载的同时减少关键边的通信开销，从而降低整个任务拓扑执行的响应时间，为任务调度策略的设计与实现提供理论依据。

定义1 有向无环图(DAG)。定义有向无环图G=(V(G),E(G)), 其中V(G)={v1,v2,…,vn} 是拓扑中的节点集合，E(G)={e,e,…,e} 是拓扑中的有向边集合。如果存在e∈E(G)，vk,vj∈V(G)，且vk≠vj，那么表示数据流从顶点vk流向顶点vj。有向无环图描述了数据流的计算过程，其中圆形为计算节点，箭头表示数据的流动方向。有向无环图如图1所示。

图1 拓扑有向无环图

定义2 关联度(association degree)。对于任意节点vk，若存在以节点vk为弧尾的n条有向边e和以vk为弧头的m条有向边e，则称弧vk→vp和弧vq→vk为节点vk的关联度AD(e,e)。

节点vk的入度(in-degree)为指向vk的有向边条数，记为ID(vi)。出度(out-degree)为从vk指向其它节点的有向边条数，记为OD(vi)。例如图1中的拓扑有向无环图中，存在顶点集合V={va,vb,vc,vd,ve,vf,vg}，有向边集E={e,e,…,e,e}，源点为va，汇点为vg。则节点ve的入度为2，出度为1。

定义3 带权路径(path of weight，PW)。定义路径p(vi,vj)，且∈V(G) 是有向无环图中一条从节点vi开始到节点vj结束的一条有向路径，定义路径上有向边的权重为W={w1,w2,…,wn}。 如果∃k，e∈p(vi,vj)，e∈p(vi,vj)，对于路径p(vi,vj) 中的任何有向边e，当k≠u时，∃x，e∈p(vi,vj)；当u≠v时，∃x，e∈p(vi,vj)。

根据流式计算的拓扑结构可知，在DAG拓扑中从源点va到汇点vg存在m条路径，既P={p1(va,vg),p2(va,vg),…,pm(va,vg)}。 如果将节点vi流向节点vk的数据流大小作为弧vi→vk的权重，那么可以将流式计算拓扑图转为带权值的AoE-网。

AoE-网拓扑模型如图2所示。

图2 AoE-网拓扑模型

由上可知，每条路径上的计算延迟是由节点的计算开销和节点间的通信开销共同决定的，因此

(1)

其中，cvi表示某条路径上节点的计算延迟，cej表示节点间的通信延迟。

那么关键路径可以表示为

Dpj=max{dp1(vs,vt),dp2(vs,vt),…,dpm(vs,vt)}

(2)

根据定义1、定义2、定义3可知，拓扑中顶点的最早发生时间ve(i)，顶点的最晚发生时间vl(j)；在拓扑有向边e上，存在最早开始时间e(k)和最晚开始时间l(k)，当e(i)=l(i) 时，表示这条有向边e在关键路径上，并且该有向边e的权值为w(e)，表示任务持续时间。

因此，按照拓扑顺序，拓扑中顶点的最早发生时间ve(i)为

(3)

其中，s为源点，源点的最早发生时间为零；E(k)是从节点i到达节点j的所有有向边的集合。

当按照逆拓扑顺序时，拓扑中顶点的最晚发生时间vl(j)为

(4)

其中，t为汇点，汇点的最晚发生时间和最早发生时间相等；E(k)是从节点i发出的所有有向边的集合。

拓扑中每条有向边e的最早开始时间e(k)为

e(k)=ve(j)

(5)

拓扑中有向边e的最晚开始时间l(k)为

l(k)=vl(j)-w(e)

(6)

算法1：关键路径检测算法(CP-Algorithm)。

输入：有向无环图G=，

有向边权值集合W←{w1,w2,…,wn}；

输出：关键节点集合Vcp，关键边集合Ecp；

(1) if ID(vi)=0；/*从源点s出发进行遍历*/

(2) for i=1;i≤n-1;i++；

(3) ve[i+1]←ve[i]+wi；/*计算vertex的最早发生时间*/

(4) vl[n-1]←ve[n-1]；

(5) if OD(vi)=0；/*从汇点t出发进行遍历*/

(6) for j=n-2;j>1;j--；

(7) vl[i-1]←vl[i]-wi；/*计算vertex的最晚发生时间*/

(8)通过式(3)和式(4)计算最早开始时间e和最晚开始时间l；

(9)when e=l；/*判断当前有向边为关键边*/

(11) end；

在算法1中，对于Flink环境中的DAG将数据流大小作为有向边的权重构建AoE-网，然后CP-Algorithm依次对数据拓扑AoE网进行正向和反向遍历，通过步骤(2)～步骤 (9)确定数据拓扑中关键节点和关键边，因此，该算法的时间复杂度为T(n)=O(n+e)，且在空间复杂度上，DAG拓扑结构并未发生改变，因此，该算法是可行的。

3 关键路径模型定义及算法

在本章节中通过算法1中检测到的关键节点集合和关键边集合对问题进行定义和建模。

3.1 关键节点负载均衡模型

(7)

(8)

(9)

由上可知，式(7)表示理想状态下关键节点的负载情况，式(9)表示关键节点的实际权重与理想权重的偏离程度，并且标准差越小表示各个工作节点的负载偏离度越低，负载越趋于均衡。

3.2 关键边最优通信开销模型

图3 任务分配模型

如上所述，在关键路径上存在节点间通信和节点内通信，且节点间通信开销远大于节点内通信开销，通过将节点间的通信开销尽可能地转化为节点内通信开销，能够降低关键路径响应时间，从而降低整个任务拓扑的响应时间。基于以上思想，提出定理1。

定理1 最优通信开销定理。当关键路径上不存在或节点间通信开销最少时，最小化的节点间通信开销等价于最大化的节点内通信开销。即

(10)

证明：由Flink拓扑模型可知，当提交拓扑给节点后，拓扑实例便不会发生改变，其包含的任务总数和数据流总数不可改变。因此，设总数据流大小为定值R，即

(11)

证毕。

对于节点内线程间通信开销，Flink提供SlotSharingGroup类，会尽可能地让更多的子任务共享一个任务槽；提供ColocationGroup类可将子任务强制放入一个任务槽内，SlotSharingGroup类和ColocationGroup类这两种方法为我们减少关键节点内线程间通信开销提供了帮助。对于节点间通信，通过Flink提供的operator chains，会尽可能地将operator的子任务chain在一起形成一个任务，每个任务在一个线程中执行，通过设置operator chains能够减少进程之间的切换，减少进程之间通信开销。因此，对于节点间通信，为达到定理1关键边最优通信开销模型要求，尽可能地将节点间通信转为节点内通信方式，并且在降低关键节点计算开销的同时降低关键边的通信开销，即在保证关键节点负载差异较小的同时降低关键节点间的通信开销，尽可能地将负载过高关键节点上的任务调度到负载较低的计算节点上。

3.3 基于拓扑关键路径的任务调度算法

基于关键路径的任务调度算法主要是在保证系统性能不发生改变的情况下，尽可能使得各关键节点负载差异较小的同时减少关键边的通信开销，从而降低整个任务拓扑执行响应时间，实现资源最大化利用。在上一节中通过关键路径检测算法确定拓扑关键路径，并获取关键路径上权重集合Wcp、节点集合Vcp、边集合Ecp。并且通过负载模型判断出负载较高的关键节点，对此负载较高节点上存在节点间通信的任务执行任务迁移策略，将该任务的节点间通信开销转为节点内通信开销，在保证关键节点负载差异较小的同时降低任务的通信开销。

为了达到上述关键节点负载均衡模型和关键边最优通信开销模型的要求，提出了一种在Flink环境下的任务调度策略(TSS-Flink)。其算法具体过程如下：

算法2：拓扑关键路径上任务调度算法。

(1)quicksort(Wcp，DESC)；

/* 对输入的关键边权重集合元素降序排序 */

(3)calculate theδby (9)；

/* 判断关键路径上是否存在负载不均衡的节点 */

/* 确定不均衡节点以及该节点上任务和前驱任务的集合 */

/* 确定关键节点上任务和它的前驱任务 */

(8) if np≠nq

(12) reschedule CP-Algorithm；

(14)end while；

4 实验与分析

Apache Flink 作为开源免费的分布式数据流处理平台之一，在实时业务中得到广泛应用。对于本章节的实验，通过在Flink平台上实现TSS-Flink策略，对该策略的有效性进行验证。

4.1 实验环境

实验环境是由7个相同配置的普通物理PC机组成的Flink集群，其中包含1个JobManager节点，该节点负责Flink集群的作业调度和资源管理；6个TaskManager节点，负责执行具体任务计划。此外，配置1个Zookeeper节点负责在任务执行过程中监控和记录数据节点；1个Kafka节点和1个HDFS节点作为数据流的源点和汇点。各节点的具体的分布情况见表1。

表1 Flink集群节点分布

在Flink集群环境中，为保证实验的顺利进行，集群均采用相同的配置，具体配置参数见表2。

为了使TSS-Flink算法达到最优的执行效果，通过多次对原系统反复实验，最终确定实验相关参数。配置情况见表3。

在本章节的实验测试中，执行了Streaming Benchmark 中的WordCount、TwitterSentiment基准测试进行验证。在WordCount基准测试中，以英文小说《Harry Potter》作为

表2 节点配置信息

表3 TSS-Flink算法参数设置

输入数据源，统计单词频次，其计算复杂度相对较低但对CPU资源的占用率较高。TwitterSentiment是一个针对Twitter用户所发的推文内容进行情感分析的作业，该作业以160 000条文本作为输入数据源，其对内存资源和CPU资源占用相对都较高。通过以上两个基准测试，能够对TSS-Flink算法的有效性进行验证。

4.2 实验分析

本章节中通过执行WordCount和TwitterSentiment这两组资源敏感型基准测试，从计算延迟、CPU负载和RAM占用率3个方面对Flink集群中各个工作节点进行性能监测和评估，以验证TSS-Flink的优化效果。

本节讨论基准测试WordCount作业在Apache Flink默认调度算法和TSS-Flink下分别运行时集群各工作节点的负载情况。由于Flink默认调度算法采用随机的方式分配任务，当从Source operator发送数据流到Sink operator时，极易导致各工作节点资源分配不均、负载差异较大情况，且TSS-Flink算法在执行过程中应该考虑到任务分配所导致的负载差异性。从图4所示的实验结果中可以得出：在Flink默认调度算法下，各个节点的CPU负载不均衡且差异较大，其中负载最高的节点是node5，负载最低的节点是node6，节点之间CPU负载最大相差28%。当节点node2和节点node5的CPU负载超过表3中设置的阈值0.7时触发TSS-Flink算法，该算法执行后集群中各工作节点的CPU负载差异明显缩小且均低于用户设置阈值0.7，且其执行后的CPU负载标准差比Flink默认调度算法降低了8.28%。通过对集群各工作节点的负载均衡测试验证了TSS-Flink算法的有效性。

图4 WordCount CPU负载对比

为了进一步验证TSS-Flink策略的优化效果，在本章节中继续对benchmark作业执行过程实时监控节点的内存占用率。在Flink中，通过Monitor模块进行内存实时监控，定义OperatorScopeFormat.java类获取System Metrics。在实时监控过程中，通过定点采样得到如图5所示的实验结果：当单位时间内数据元组数量不断增加时，原系统部分节点由于负载过高导致内存占用率急剧上升，并且这些负载较高的节点无法及时处理数据从而导致拓扑处理时延变长，而另外一部分节点的资源也无法得到充分利用。通过使用TSS-Flink策略，对负载较高的节点上的任务重调度，使得拓扑非关键路径上的节点分担拓扑关键路径上负载过高节点的资源使用压力，最终被采样节点的内存利用率都有一定程度的下降且逐步趋于平稳状态。

图5 WordCount内存占用对比

图6表示benchmark在Flink默认调度算法和TSS-Flink下的工作节点间通信开销，不管是在默认调度算法还是TSS-Flink下，节点间数据流大小均经历一个从0快速上升到正常状态的过程。TSS-Flink算法在执行中将关键节点上的线程迁移至前驱非关键节点上，从而减少线程节点间通信开销。Flink默认调度算法运行且趋于稳定后(90 s-300 s)，节点间数据流大小的平均值约为16 572 tuples/s；当执行TSS-Flink算法且系统趋于稳定后(125 s-250 s)，节点间数据流大小约为12 410 tuples/s，相比Flink默认调度算法降低了25.1%。可见，TSS-Flink在降低节点间通信开销方面具有更为明显的效果且符合最优通信开销模型思想，也进一步验证了算法的有效性。

图6 节点间数据流大小对比

图7表示任务拓扑中汇点接收从source发出的每 10 000 条tuples时记录一个延迟时间并持续15 min得到的实验结果，实验在WordCount作业上执行TSS算法与原系统算法比较得出：因为WordCount作业复杂度低于Twitter作业复杂度，所以WordCount作业的计算延迟相对较低，当经过TSS-Flink算法进行优化后，系统的计算延迟明显下降。在原系统中，随着数据流的连续不断输入，计算延迟也随着慢慢升高，当某些节点的计算资源达到瓶颈无法及时处理数据时，导致计算延迟过长，系统执行任务拓扑的实时性较差。通过TSS-Flink算法，对计算资源相对紧张的关键节点上的任务执行调度策略，对比原系统该策略使集群的计算延迟降低最多达到388 ms，至少40 ms，平均降低了248 ms。

图7 WordCount计算延迟对比

对于图8，执行的Twitter作业本身复杂度比WordCount作业较高，因此计算延迟也较高。通过在连续15 min内记录汇点每接受10 000条tuples数据时的延迟时间，可以得出：原系统在数据流不断增加和快速变化下，导致数据流无法及时处理，造成数据堆积，数据堆积节点延迟过高，进而影响系统的实时计算能力。通过执行TSS-Flink算法将任务拓扑中每10 000条tuples数据的计算延迟最多降低了210 ms，至少降低了8 ms，平均降低了130 ms，该调度策略有效地降低了节点间通信开销和关键路径上计算延迟，提高了计算的实时性，使计算资源达到最大化利用。

图8 Twitter计算延迟对比

综上所述，实验验证TSS-Flink算法对够通过降低节点间通信开销从而降低响应时间，提高集群的性能。通过图7、图8可知，不同的作业类型下该策略对系统的计算延迟优化效果并不相同，但其平均优化比提高了13.09%，有效地降低了计算延迟，提高了系统性能。

5 结束语

通过对比现有的任务调度算法，发现多是对负载较重的节点执行任务调度策略，虽然这些调度策略能有效降低任务拓扑响应时间，提高系统性能。但也并未考虑到各任务的计算开销和任务之间的通信开销，且并未在Apache Flink平台上实现该任务调度策略，所以在节点间负载均衡和通信开销方面仍然存在很大的优化空间。本文通过找到直接影响整个任务拓扑响应时间的关键路径，确定负载较高的关键节点和该节点上通信开销较大的关键边，建立关键节点负载均衡模型和关键边最优通信开销模型，提出一种Flink环境下的任务调度策略(TSS-Flink)。通过WordCount和Twitter两个benchmark的实验验证，结果表明算法能够实现对Flink集群的性能优化，尽可能地更好地利用计算资源。

下一步的研究工作将重点关注由于输入数据流的急剧变化造成的资源分配不均问题，针对关键路径上的负载倾斜较为严重的关键节点，如何判断出节点内的任务通过横向迁移和纵向迁移实现资源的最大化利用且保证迁移后的拓扑结构不发生改变。