融合深度强化学习与算子优化的流式任务调度

摘要： 针对实时性要求高和作业量大的流处理作业执行过程中，多个作业之间存在的相同处理片段可能会导致流处理引擎重复计算、资源浪费和处理性能低下的问题，提出了融合深度强化学习与算子优化的流式任务调度方法。首先利用算子优化算法将多个复杂的作业去重、重构，其次将重构得到的作业输入循环神经网络中得到任务的调度策略，最后利用强化学习模型进行调度策略的优化。所提方法利用算子优化减少了每个作业中创建的算子实例，结合深度强化学习自动发现最优的调度策略，有效地避免了因大量实例运行而造成的系统资源不足、数据拥塞等问题。对比实验结果表明，所提方法在吞吐量和延迟方面的表现更优异。

关键词： 流处理作业； 任务调度； 算子优化； 深度强化学习

中图分类号： TP18

文献标志码： A

文章编号： 1671-6841（2025）01-0015-08

DOI： 10.13705/j.issn.1671-6841.2023159

Stream Processing Task Scheduling Integrating Deep Reinforcement

Learning and Operator Optimization

GUO Chenhong^1，2， WANG Jing^1，2， GONG Huilong^1，2， GUO Haohao^1，2， ZHANG Ruixuan^1，2

（1.School of Information， North China University of Technology， Beijing 100144， China;

2.Beijing Key Laboratory of Large-scale Streaming Data Integration and Analysis Technology

（North China University of Technology）， Beijing 100144， China）

Abstract： Aiming at the problems of high real-time requirements and large workload during the execution of stream processing jobs， the same processing fragments among multiple jobs might lead to repeated calculations of stream processing engines， waste of resources， and low processing performance， a scheduling method for stream processing tasks that integrated deep reinforcement learning and operator optimization was proposed. Firstly， the operator optimization algorithm was used to deduplicate and reconstruct multiple complex tasks. Secondly， the reconstructed tasks were fUkXwAsx4LNglHN0AoE+wQ==input into a recurrent neural network to obtain the scheduling strategy for the tasks. Finally， the scheduling strategy was further optimized using a reinforcement learning model. The proposed method reduced the number of operator instances created in each task through operator optimization. By combining deep reinforcement learning， the method automatically discovered the optimal scheduling strategy， effectively avoiding issues such as insufficient system resources and data congestion caused by a large number of instances running. The comparative experimental results showed that the proposed method performed better in terms of throughput and latency.

Key words： stream processing job; task scheduling; operator optimization; deep reinforcement learning

0引言

流处理引擎可以从连续产生的数据流中实时地提取有用的信息，具有较快的响应速度和更好的实时性。然而，面对大量的实时产生的数据，流处理引擎需要在硬件设备等其他限制条件下快速地处理数据^［1］。因此，提高流处理的性能和效率迫在眉睫。提高其性能可以考虑优化流处理作业中任务调度策略的方式^［2-4］，然而确定任务调度策略是一个NP难的问题。

目前，已有一些启发式和深度强化学习的算法对如何确定任务调度策略进行了优化，考虑了流处理作业中算子的特征、计算资源同构或者异构等多种影响因素，尽可能地优化任务的调度执行。其中，算子是作业中的基本操作单元，而任务是由多个算子按照一定的顺序和依赖关系组合而成的可执行单元。但是，确定任务调度策略还存在一些可以改善的地方。例如在大规模流数据到来之后，待处理的作业量很庞大，需要为一些小的作业单独创建多个作业实例，一定程度上消耗了过多的资源。同时，这些计算作业之间有很多重复的算子组合片段，导致了处理引擎的重复计算，浪费了计算资源。

相较于一些传统的启发式算法，深度强化学习算法^［5］可以动态地调整任务调度策略，不需要预先指定规则即可发现最优调度策略，并且能灵活设置优化目标。本文提出一种融合深度强化学习与算子优化的流式任务调度方法（stream processing task scheduling integrating deep reinforcement learning and operator optimization），简称为DRLAOO。该方法首先采用有向无环图表示流处理作业的任务执行图，采用连通图表示计算资源的特征。其次，利用图卷积神经网络提取作业和计算资源的特征，把这些特征输入循环神经网络（recurrent neural network，RNN）中，利用注意力机制得到每个作业中任务的调度结果。最后，利用强化学习模型优化任务的调度策略。在对比实验中，DRLAOO方法获得了较低的延迟和较大的吞吐量，节约了计算资源的使用。

1相关工作

1.1任务调度级别优化

在任务调度算法方面，有传统的调度算法以及深度强化学习算法。Agrawal等^［6］提出一个基于度量的并行实时任务模型，并针对相应的实时调度问题，提出一种基于任务最大响应时间的调度策略。Blamey等^［7］研究了混合云/边缘部署设置中对CPU成本和消息大小敏感的调度流处理，目的是在有限的边缘资源下最大限度地提高吞吐量。Li等^［8］研究了3种调度策略，其中有关基于分解的并行任务调度是把多核机器上顺序任务开发的调度算法作为黑盒，每个并行任务被分解为一组顺序任务，然后使用一些已知的调度器对多核机器上的顺序任务进行调度。Ni等^［9］研究了在分布式系统中对连续数据流进行实时处理的流处理中的资源分配问题，提出一个图感知编码器-解码器框架来学习一种广义的资源分配策略。Huang等^［10］在此研究基础上进行了优化，在计算资源和网络资源是异构的情况下设计了一个通用的基于深度强化学习的资源感知框架，利用图嵌入和注意力机制对资源进行建模，预测任务分配的位置。但是，上述这些工作关注于单个任务级别的优化调度，很少考虑作业的复杂性和作业之间的算子重复片段导致的流处理引擎的重复计算问题。

1.2算子级别优化

王亚等^［11］提出了基于匹配结果共享的方法来优化流数据的处理，利用已经计算出结果序列的共享子序列来构造新的结果序列。该方法不必对结果序列的每个候选算子实例都进行计算，在一定程度上提高了计算的效率。Bok等^［12］考虑了对来自传感器的流数据的相似操作和冗余操作，将作业中的相似算子转换为虚拟算子，将相同任务中的冗余算子转换为单个算子。使用转换后的算子重构作业，通过这种方法降低了相似算子和冗余算子的计算成本，从而降低了整体处理成本。He等^［13］提出了划分和聚类算法来优化流数据的处理，针对作业中存在的相同作业处理片段，用划分算法消除一些重复，减少了重复计算。Higashino等^［14］提出一种属性图重写AGeCEP方法，针对复杂事件处理任务中的算子问题，提出了算子共享的概念，即单个算子实例被两次或多次出现的算子共享的能力。上述这些工作考虑了算子级别优化，针对的是单个作业的优化，未考虑流处理引擎执行过程中的任务调度问题。

2任务调度方法DRLAOO

2.1定义

定义1流处理作业任务执行图G。G用于表示作业中各个算子之间的依赖关系，定义G=（V，E），其中：V表示将要被分配到计算资源上的算子；E表示算子V之间传递的数据流。

定义2算子V。V是流处理作业的基本功能单元，从输入消耗数据，对它们进行计算后输出数据做下一步的处理，其中：V=（id，operator_type，cpu，memory，is_source，is_sink）。

定义3异构设备资源图G。一个G表示一组计算资源的设备，使用全连通图G=（V，E）表示，其中：V表示放置任务的最小单位。

2.2任务调度方法原理

融合深度强化学习与算子优化的流式任务调度是一个用于优化流处理作业执行的调度方法。任务调度方法原理图如图1所示。

该方法的基本思路是将多个作业抽象为SET<G>，利用算子优化算法将多个作业去重、重构。处理之后的图命名为G，G不改变流处理作业的数据源和处理逻辑等信息，保证去重、重构之后的数据源和处理逻辑是一样的。

将G和G输入深度强化学习的任务调度模型中，深度学习模型根据输入的作业和计算资源的特征信息得到任务调度策略，强化学习模型根据得到的奖励值不断地更新深度学习模型的参数，最终得到较优的任务调度策略P。

该任务调度方法由以下2个部分组成。

1）算子优化算法。该模块用于调度前对多个作业进行处理。当SET<G>到来之后，利用算子优化算法将其进行去重、重构，最终得到G参与后续的任务调度阶段。

需要强调的是，经过算子优化算法得到的G包含了去重优化之前的所有处理逻辑。

2）任务调度阶段。该任务调度阶段由GraphSage网络、RNN网络以及强化学习模型组成。组合调度算法的输入是上一步得到的G以及G，这些算法模型的组合运作方式如下。

① 利用GraphSage网络提取作业和计算资源中的图嵌入和节点嵌入的信息，例如，图嵌入捕捉了V之间的整体结构和关系，节点嵌入捕捉了每个V的特征属性。

② 将提取到的信息输入RNN中，其中向前传播的隐藏状态可以用来存储当前已经执行的V的调度策略。将V的调度策略输入RNN中进行训练，得到相应的调度策略以及每个调度策略所对应的吞吐量、延迟、CPU和内存。

③ QoS属性估计：将吞吐量、延迟、CPU和内存作为调度策略的评价指标，估计每个任务调度策略对应的仿真情况下的吞吐量、延迟、CPU和内存。

④ 以评价指标作为强化学习算法的奖励来优化任务调度策略，此算法的输入是上述的RNN以及每个任务调度策略对应的QoS属性估计值。采用策略梯度算法进行训练，更新RNN的参数，得到较优的任务调度策略P。

2.3算子优化算法

基于文献［14］中的Processing+Source算子实例共享策略，提出了算子优化算法。该算法中的算子实例共享策略表示为输入源相同以及执行相同的数据处理情况下的共享算子实例。

举例说明，假设有2个作业都包括4个算子：数据源、数据过滤、映射和输出算子。它们的处理过程非常相似，只是在输出算子上不同。作业1的输出算子是File，作业2的输出算子是Kafka。利用算子优化算法对这2个作业进行算子合并、重构的过程包括：遍历每个作业中的所有算子；作业的合并和重构。

优化前的作业示例如图2所示。2个作业除了输出算子，其他算子都是相同的，在这样的情况下需要创建2个作业实例。经过算子优化处理之后的作业示例如图3所示。

将作业优化之后，只需要创建1个作业实例即可。如果存在大量的相互之间有重复片段的作业，利用算子优化算法可以发现多个作业之间相同的算子组合片段，将它们合并、重构之后再参与后续的处理。算子优化算法遍历了每一个任务执行图，迭代地将任务执行图两两合并。合并方式如下。

第1步判断算子类型是否为源算子。如果是源算子且两者类型相同，则合并为一个源算子；

如果不相同，则保留两个源算子。

第2步遍历处理算子。如果两个任务执行图中算子片段相等，则合并类型相同的算子；如果不完全相同，则只合并相同的算子，增加分支构造图即可。

第3步判断输出算子类型是否相同。相同则合并，不相同不进行合并，而是保留多个输出算子。

该算法的具体步骤如下。

算法1算子优化算法（OOA）

输入： 多个流作业任务执行图SET<G>。

输出： 算子优化之后的任务执行图G。

1． while len（SET<G>）>1

2．for i in range（0， len（SET<G>），2）

3．if i+1 < len（SET<G>） then

4．mergedGraph=mergeGraphs（G，G）

5．G=mergedGraph

6．SET<G>.pop（i+1）

7． end if

8．end for

9． end while

算法2合并流处理作业任务执行图算法mergeGraphs

输入： 两个任务执行图G，G。

输出： 经过算法mergeGraphs合并后的任务执行图mergedGraph。

1． mergedGraph=Graph（）

2． mergedGraph.sources=mergeSources（G.sources， G.sources）

3． mergedGraph.operators=mergeOperators（G.operators，G.opertors）

4． mergedGraph.sinks=mergeSinks（G.sinks， G.sinks）

5． return mergedGraph

图4是一个算子优化示例，该例子简单描述了数据源和源节点类型相同的4个作业的合并、重构优化的过程。

2.4深度强化学习任务调度算法

作业被重构优化之后得到的G，将参与后续的任务调度。图5展示了深度强化学习模型架构。

该模型由GraphSage网络提取G中V以及G的特征信息，交由RNN生成每个V所对应的任务调度策略P。同时，强化学习模型使用训练期间观察到的奖励对神经网络参数执行梯度下降来更新RNN的参数。

2.4.1GraphSage网络

GraphSage是图卷积神经网络中的一种，可以在大规模图上学习图嵌入和节点嵌入的信息^［15］。

GraphSage网络首

先对邻居节点随机采样，然后生成目标节点嵌入，最后将节点嵌入作为全连接层的输入，预测目标节点的标签。具体公式表示为

h^k=mean（h^k-1，u∈N（v）），

h^k=σ（w^k·Concat

（h^k-1，h^k）），（1）

式中：k表示当前层数；h^k-1是第k-1层中节点u的向量表示。将节点v在k-1层的多个邻居节点通过mean函数（即平均池化）得到节点v的邻居表示h^k。然后将h^k和该节点在第k-1层的表示h^k-1进行拼接，进而通过一个全连接神经网络（即w^k）得到节点v在第k层的向量表示h^k。

将G、G经过预处理之后输入GraphSage网络中，得到V、V的嵌入向量分别定义为emb、emb。

2.4.2RNN

使用RNN来跟踪计算过程中的状态^［16］，将每个算子V分配到V上，当前节点的分配策略P由G、G和所有上游节点分配的资源节点的情况V^up计算得到。即

PG，G=

∏

ΠiP（Vⁱ

V^up，G，G）。

（2）

结合RNN的注意力机制，通过学习状态表示S来记忆依赖项，编码与V^up和G相关的信息，在每一步中，输入向量emb将在V^up中添加新的资源节点的嵌入特征emb，

S=LSTM_Cell（emb，S，V^up）。（3）

每个算子的调度策略将根据当前状态和任务的相关信息计算得到，重复执行该过程，模型最终得到所有算子对应的调度策略P。

2.4.3强化学习Policy-Gradient网络

使用策略梯度函数进行训练^［17-18］，根据训练期间的奖励对神经网络参数执行梯度下降，不断地更新RNN模型的参数。将所有参数表示为θ，所有可能的放置方案的分布表示为π，模型的最大化目标为

J（θ）=∑pπ（P）r（P），（4）

式中：r表示奖励，范围为0～1。

接着使用REINFORCE算法计算策略梯度，学习网络参数θ，

ΔJ（θ）=1N∑Nn=1

Δlogπ（P）［r（P）-b］，（5）

式中：b为N个样本的平均奖励，用于降低策略梯度的方差；Δθlogπ（P）提供了一个方向来增加选择P的概率。

将学习率表示为α，参数θ通过θ←θ+αΔJ（θ）进行更新，可以增加选择到奖励高于平均水平的调度策略P的概率。

3实验与分析

3.1实验准备

3.1.1数据集

为了验证本文提出的融合深度强化学习与算子优化的流式任务调度方法DRLAOO的有效性，在文献［10］的基础上改造了表示作业的数据集，加入了算子类型的特征属性，更适用于所提出的问题场景。

结合表示计算资源特征的数据集构建了相应的训练集、测试集和验证集。其中，1个表示作业的任务执行图和1个表示一组计算资源的图构建为1个数据集样本，构建完的数据集样本共有1 500个。

实验用到的流处理作业数据集和计算资源数据集描述见表1和表2。

3.1.2实验设备

实验部分采用了以下2台设备。

设备1： 用来训练构建好的算法模型，型号为CentOS6，32 GB。

设备2： 用来构建算法模型以及加载设备1上训练好的模型，并验证模型的输出结果，型号为Windows11，16 GB。

2台设备上安装的第三方库版本为Python3.8.16，Pytorch1.8.0。

3.2对比实验

选取Flink、Storm流处理引擎默认的调度方式以及吞吐量感知的任务放置框架（TATA）^［10］进行对比实验。

Flink默认的调度策略：按照数据源的分布情况优先考虑将任务调度到距离其最近的执行器上执行，以利用本地计算资源和缓存减少数据传输和网络开销。但是，在计算资源节点性能相差较大的异构集群中，该策略会导致一些计算资源较弱的节点长时间运行任务，带来局部负载不均衡的问题。

Storm默认的调度策略：基于轮询的机制，将任务均衡地分配到可用的计算资源上，每个计算资源在同一时间只处理1个任务。这种调度策略适用于较小规模的集群和任务。

吞吐量感知的任务放置框架（TATA）：综合考虑作业中任务的特征和异构计算资源的特征，使得任务合理地分配到相应的计算资源节点上，但是不太适合具有大规模任务的流处理作业。

3.3结果分析

图6和图7分别比较了DRLAOO方法与其他3种调度策略在不同作业个数下的延迟和吞吐量。

从图6可以看出，随着作业个数的增多，4种调度策略下作业的延迟不断增加；DRLAOO方法在总体上相较于其他调度策略的延迟较低。Flink默认的调度策略是将相关的上下游任务放置在相同的计算资源中;Storm默认的调度策略是将任务平均分配给集群内的所有主机，未考虑任务之间的依赖关系;TATA将简单相关的任务放在计算资源少的位置上。而DRLAOO方法是对基于TATA的工作进行了改进，提出的算子优化算法达到一种算子共享的效果，减少了创建的算子实例个数，因此具有较低的延迟。

从图7可以看出，随着作业个数的增多，4种调度策略下的吞吐量在不断增加后趋于平稳。这是因为计算资源的限制和网络带宽的限制等因素会导致吞吐量的增长趋于缓慢甚至停止增长，最终趋于平稳。DRLAOO方法在一定程度上减少了作业之间相同的处理片段造成的流处理引擎重复计算的问题，节约了资源的使用，进而给其他任务预留了资源，增大了吞吐量。

图8和图9分别比较了10个作业参与执行，不同重复率下DRLAOO方法与TATA的延迟和吞吐量。

由图8和图9可知，当多个作业之间的重复率为0时，2种调度策略下的延迟和吞吐量相差不大。这是因为当重复率为0时，用算子优化算法对多个作业进行优化，由于不存在可以合并的算子，所以未能减少需要创建的作业实例的个数，因此不能在延迟和吞吐量的指标上有更好的表现。

在不同作业个数下的CPU使用情况和内存使用情况。由于是在仿真环境，所以数值仅供参考，但是可以看出大致的趋势是DRLAOO方法将会消耗更少的资源。

4结语

本文提出了一个融合深度强化学习与算子优化的流式任务调度方法DRLAOO，该方法针对大量的有重复片段作业的情况，可以尽可能地利用算子优化减少每个作业中需要创建的算子实例，结合深度强化学习自动发现最优的调度策略，进而获得了较低的延迟和较大的吞吐量，提高了处理效率，节约了计算资源的使用。但是，本研究还存在一些可以改进的地方：一是利用算子优化时可考虑更多的特征，例如算子之间的通信延迟。同时，考虑对算子优化算法进一步优化以处理拥有多个数据源的作业的合并。二是将考虑结合图论，采用更贴合实际的方法来表示异构资源，使得调度策略更准确。

参考文献：

［1］DONGEN G， DEN P D. Influencing factors in the scalability of distributed stream processing jobs［J］. IEEE access， 2021， 9： 109413-109431.

［2］LI C L， LIU J， LI W G， et al. Adaptive priority-based data placement and multi-task scheduling in geo-distributed cloud systems［J］. Knowledge-based systems， 2021， 224： 107050.

［3］CHEN L S， WU D P， LI Z D. Multi-task mapping and resource allocation mechanism in software defined sensor networks［C］∥International Conference on Wireless Communications and Signal Processing. Piscataway：IEEE Press， 2020： 32-37.

［4］GELDENHUYS M K， SCHEINERT D， KAO O， et al. Phoebe： QoS-aware distributed stream processing through anticipating dynamic workloads［C］∥IEEE International Conference on Web Services. Piscataway：IEEE Press， 2022： 198-207.

［5］刘全， 翟建伟， 章宗长， 等. 深度强化学习综述［J］. 计算机学报， 2018， 41（1）： 1-27.

LIU Q， ZHAI J W， ZHANG Z Z， et al. A survey on deep reinforcement learning［J］. Chinese journal of computers， 2018， 41（1）： 1-27.

［6］AGRAWAL K， BARUAH S， EKBERG P， et al. Optimal scheduling of measurement-based parallel real-time tasks［J］. Real-time systems， 2020， 56（3）： 247-253.

［7］BLAMEY B， SINTORN I M， HELLANDER A， et al. Resource-and message size-aware scheduling of stream processing at the edge with application to realtime microscopy［EB/OL］.（2019-12-19）［2022-12-23］. https：∥doi.org/10.48550/arXiv.1912.09088.

［8］LI J， AGRAWAL K， LU C Y. Parallel real-time scheduling［M］∥Handbook of Real-time Computing. Berlin： Springer Press， 2022： 447-467.

［9］NI X A， LI J， YU M， et al. Generalizable resource allocation in stream processing via deep reinforcement learning［C］∥Proceedings of the AAAI Conference on Artificial Intelligence. Palo Alto： AAAI Press， 2020： 857-864.

［10］HUANG X， JIANG Y， FAN H， et al. TATA： throughput-aware task placement in heterogeneous stream processing with deep reinforcement learning［C］∥IEEE International Conference on Parallel & Distributed Processing with Applications， Big Data & Cloud Computing， Sustainable Computing & Communications， Social Computing & Networking. Piscataway： IEEE Press， 2021： 44-54.

［11］王亚， 孟耀伟. 基于匹配结果共享的复杂事件检测方法［J］. 计算机应用研究， 2014， 31（8）： 2338-2341.

WANG Y， MENG Y W. Method of complex events detection based on shared matching results［J］. Application research of computers， 2014， 31（8）： 2338-2341.

［12］BOK K， KIM D， YOO J. Complex event processing for sensor stream data［J］. Sensors， 2018， 18（9）： 3084.

［13］HE S Q， CHENG B， HUANG Y Z， et al. Proactive personalized services in large-scale IoT-based healthcare application［C］∥IEEE International Conference on Web Services. Piscataway：IEEE Press， 2017： 808-813.

［14］HIGASHINO W A， EICHLER C， CAPRETZ M A M， et al. Attributed graph rewriting for complex event processing self-management［J］. ACM transactions on autonomous and adaptive systems， 2016， 11（3）： 1-39.

［15］白铂， 刘玉婷， 马驰骋， 等. 图神经网络［J］. 中国科学： 数学， 2020， 50（3）： 367-384.

BAI B， LIU Y T， MA C C， et al. Graph neural network［J］. Scientia sinica： mathematica， 2020， 50（3）： 367-384.

［16］SHERSTINSKY A. Fundamentals of recurrent neural network （RNN） and long short-term memory （LSTM） network［J］. Physica D： nonlinear phenomena， 2020， 404： 132306.

［17］刘建伟， 高峰， 罗雄麟. 基于值函数和策略梯度的深度强化学习综述［J］. 计算机学报， 2019， 42（6）： 1406-1438.

LIU J W， GAO F， LUO X L. Survey of deep reinforcement learning based on value function and policy gradient［J］. Chinese journal of computers， 2019， 42（6）： 1406-1438.

［18］赵沛尧，黄蔚.基于动态优先级的奖励优化模型［J］.郑州大学学报（理学版），2022，54（1）：62-68.

ZHAO P Y， HUANG W. Constrained reward optimization with dynamic preferences［J］. Journal of Zhengzhou university （natural science edition）， 2022， 54（1）： 62-68.