基于时间序列模型的Kafka 系统智能化管理方法①

周宇泽 司鹏搏 张延华 李 萌 杨睿哲

(北京工业大学信息学部 北京 100124)

0 引言

近年来,随着互联网普及程度的不断提高和通信技术的高速发展,信息数据的安全问题引起了社会的广泛关注。区块链有着高度去中心化、不可篡改的特点,具有很高的安全性。目前,在各个领域中都进行着区块链的研究,例如车辆自组网(vehicle adhoc network,VANET)[1]、物联网(Internet of Things,IoT)[2]、智慧城市[3]和云计算[4]等。共识算法帮助区块链更好地完成节点间点对点通信、完成交易。比较有代表性的共识算法包括工作量证明(proof of work,PoW)[5]、权益证明(proof of stake,PoS)[6]、空间量证明(proof of space,PoSpace)[7]、实用拜占庭容错(practical Byzantium is fault-tolerant,PBFT)[8]以及近些年来逐渐被人们所熟知的Kafka 共识算法。

Kafka 是一个高性能的分布式消息发布和订阅系统[9]。Kafka 中,Kafka Broker[9]指Kafka 集群的服务器节点[10],每个Kafka Broker 提供了名为主题的逻辑概念。主题[11]是存储消息的逻辑概念,被认为是消息的集合,同类型的消息会处于同一个主题中。生产者将消息以push 的形式发送到Kafka Broker,而消费者从Kafka Broker 消费消息的过程是pull,主动拉取数据,完成共识。

应用Kafka 算法的区块链系统经常会面对海量交易等待处理的情况[12-13],大量交易集中在少数节点的情况,会导致系统性能下降[14]。针对Kafka 系统中的消费者和分区,文献[15]和[16]分别提出消费者/客户端负载均衡方法和改进型Partition 过载优化,来解决吞吐速率下降、中央处理器(central processing unit,CPU)使用率过高等问题。文献[17]提出基于抽样的自适应调优方法,通过优化生产者个数、缓存空间大小、消息大小、分区数等系统参数来提升系统吞吐速率并降低时延。先前的研究大多致力于从消费者、分区和系统参数的角度,在算法和系统内部添加机制来进行优化,但却往往忽略了生产者。而生产者节点一般同时担任排序节点[18]、背书节点[18]和提交节点[19]的责任。因此生产者需要执行的进程十分复杂多样,大量数据集中在生产者的缓存区[20]中是Kafka 系统性能吞吐速率下降、时延上升甚至出现宕机情况的主要原因。因此从生产者的角度进行系统优化也是研究的重点。

长短期记忆网络(long short term memory,LSTM)模型[21]从本质上讲是循环神经网络(recurrent neural network,RNN)的一种特定形式[22]。LSTM 模型以RNN 模型作为基础模型,使用增加门限的方法来解决RNN 模型中的短期记忆信息持久性不够高的问题。时间序列[23]分析是将某种现象某一个统计指标在不同时间上的各个数值,按时间先后顺序排列而形成的序列,旨在预测未来事件发展的趋势和规律。相比其他网络,LSTM 的优势包括:输入及输出属于时序数据,并要求全局化处理;输入和输出的元素级别对应的时间跨度大;数据长短适中。LSTM应用的领域包括:文本生成、机器翻译、语音识别、生成图像描述、预测疾病、故障[24]和股票[25]等。

本文结合区块链节点的特点,提出基于LSTM时间序列的系统管理方法,在不过多修改Kafka 算法内部机制的条件下,使系统更加智能化。利用LSTM 时间序列模型可以胜任预测工作的特点,根据系统之前接收到的交易量,预测下一时刻即将收到的交易量,动态调整生产者的个数,既可以解决吞吐速率下降、时延上升等性能问题又不造成资源浪费,使系统资源利用处于相对合理区间。

1 问题建模

本节首先对问题及解决方案进行分析建模,而后建立整体系统优化模型。

1.1 Kafka 模型

目前常用的Kafka 系统,以吞吐速率、时延、能耗、系统效率、使用率等指标来衡量系统的性能和合理性。系统的总吞吐速率与生产者吞吐速率以及分区关系为

sumt表示t时刻需要处理的总交易量:

式中,Nat+Nbt+…+Nnt是t时刻各用户向系统发送的交易量。

其中,kt表示t时刻生产者数量,bm代表t时刻需要处理的单个消息大小,表示t时刻系统需要处理的消息总大小,Td表示时延。

理想状态下,每个生产者吞吐速率相等,由式(2)、(3)、(4)可得:

生产者、消费者、分区各占一部分系统存储空间。一般理想状态下,各生产者所占空间相等[25]。

Ω代表机器(节点)数量:

一般情况下Ω≥kt＞1,Cn表示副本数量,wt为权重系数,Vmax表示生产者峰值生产速度。

根据式(6),在理想系统状态下,系统最大时延Td可表示为

Et为每消耗单位能量处理的交易大小:

其中,E为节点作为生产者时,每秒消耗的能量。

Mt为生产者系统使用率:

Wt为系统效率:

其中,rt为比例系数,代表t时刻系统可承受交易量范围和用户发送总交易量范围之比。

1.2 LSTM 时间序列模型

找到生产者个数kt、用户发送交易量sumt与各项性能、能耗及系统使用情况的关系后,如何确定生产者个数尤为重要。构建预测模型,如图1 所示。

模型学习先前用户发送的交易量,预测t时刻各用户发送的交易量和系统需要处理的总交易量sumt。

LSTM 模型主要包含了遗忘门、输入门、输出门与1 个细胞单元。细胞单元状态可以控制信息传递给下一时刻。时间序列模型流程示意图如图2 所示。

图中,ct是细胞单元,从下方输入ft、it、ot分别为遗忘门、输入门、输出门,用sigmoid 层表示。sumt为输出,即预测到的交易总量。Nat-1,Nbt-1,…,Nnt-1为输入,即上一时刻用户发送交易量。ct-1为需要遗忘的对预测无用的数据。2 个tanh 层分别对应细胞单元的输入与输出。

1.3 LSTM 模型执行步骤

LSTM 结构中使用了2 种激活函数[26],分别为sigmoid 函数与tanh 函数,其表达式分别为

第1 步决定遗忘细胞状态中的哪些信息。

遗忘门通过sigmoid 层来筛选,只有符合预期结果的信息可以通过细胞单元,根据上一时刻的输出和当前时刻的输入产生一个0～1 之间的ft值,以此决定是否让上一时刻学习到的用户发送交易量信息通过或部分通过,0 表示完全舍弃,1 表示完全保留。

其中,ft为σ网络激活层,sumt-1为预测到上一时刻输出的交易总量,Nat-1,Nbt-1,…,Nnt-1为上一时刻输入的各用户发送交易量,Wf为遗忘门权重,bf为遗忘门偏执量。

第2 步产生需要更新的信息。

第2 步为2 个部分,第1 部分中输入门通过sigmoid 层决定哪些信息用来更新候选值,第2 部分生成新的候选值˜ct,并评估其保存至细胞状态的可能性。通过tanh 层来执行本部分,对输入端进行激活,对细胞单元执行更新操作,将ct-1更新为ct。

其中,Wi、Wc为权重,bi、bc为偏执量。

对旧的细胞状态进行更新,通过遗忘门消除不需要的信息,然后得出候选值。

第3 步决定模型输出。

通过sigmoid 层来得到一个初始输出,而后缩放ct的值至-1～1 之间,再与得到的输出进行逐对相乘,得到最终的输出。

sigmoid 函数的输出不考虑先前时刻学到的信息,tanh 函数对先前学到信息的压缩处理,起稳定数值作用,智能化系统整体模型如图3 所示。

本文构建的生产者个数决策模型,学习前(t-1) s用户发送交易的规律,通过时间序列模型LSTM 算法使用仍然适用于本文模型的sigmoid 函数与tanh函数完成预测第ts 用户发送的交易量的工作,输出下一时刻选定的生产者个数,对生产者个数进行动态调整,在保证系统性能的情况下,达到降低能耗、合理利用系统资源的要求。

2 仿真实验

仿真环境建立在英特尔11th Gen Intel(R) Core(TM)i5-1155G7@2.50 GHz 处理器。机带RAM 16.0 GB 64 位操作。软件环境是在Win 10 操作系统下运行Python 3.6、Matlab 2022 以及Jupyter Lab 平台。

分别设置模型学习率为0.01、0.05,并结合实际应用中Kafka 系统一般接收到的交易量,将样本数量设置为50 个、100 个、150 个进行对比实验。

进行500 回合的对比实验训练时,由图4 中十字标记曲线可知,当学习率设定为0.05、样本数量确定为100 个时,系统误差最小,最低可降到0.0124,收敛于0.002 左右。因此LSTM 模型学习率最终确定为0.05,样本数量选定为100,如表1 所示。

表1 LSTM 参数设置

图4 模型参数对比

后续实验中,基于时间序列LSTM 算法的Kafka优化方法设置详细步骤如算法1 及图5 所示。

图5 优化方法实验设置

交易数据集为随机生成,总体处于1～100 个,通过Excel 导入模型。模型参数设置如表2 所示。

表2 模型参数设置

时间序列模型执行步骤如下所示。

第2 节公式中n取1、2、3,分别设定3 个序列模型。Nat、Nbt、Nct分别对应成员1、成员2、成员3在第ts 预测发送的交易量。通过总交易量所处区间,决定生产者个数。根据第2 节中公式可得LSTM时间序列模型输出与Kafka 系统性能关系,如表3所示。

表3 LSTM 输出与Kafka 性能的关系

使用MinMaxScaler 数据归一化方法:

其中,X.min(axis=0) 表示每列值的最小值组成的向量,同理X.max(axis=0) 表示每列值的最大值组成的向量。max 表示映射到区间的最大值,默认为1。min 表示映射到区间的最小值,默认为0。Xstd是标准化化结果,Xsvaled是归一化结果。使用均方误差(mean square error,MSE)作为误差评价指标:

式中,N为样本数量,Yi是为真实值,为模型预测值。优化器选择适应性矩估计(adaptive moment estimation,ADAM)的方法,对损失函数进行优化。

2.1 LSTM 预测

使用数据集分别进行500、1000、3000、10 000 次训练。当训练次数达到10 000 次时,误差率曲线在0.025 以下收敛,最小可以达到(8.2807e)-4,损失函数如图6 所示。

图6 误差曲线

训练后得到预测下一时刻各用户发送的交易量,并根据交易总量决定k取值。结果如图7～9 所示,在折线图上方显示系统经过预测后选定的生产者个数。

图8 预测图2

图9 预测图3

实验中会出现k=1、k=2 和k=3 这3 种结果。图中显示选定的k值,数据中包含需要通过遗忘门消除的对预测无用数据。用户1数据波动大,分布不均匀;用户2 数据基本均匀分布在小范围内;用户3 数据波动处于用户1 与用户2 之间,分布基本均匀。通过图7～9 可以看出,对预测结果影响最大的是用户1,用户2 与用户3 对结果影响不明显。

2.2 性能对比验证

根据前述公式及表1 可知时间序列模型预测的消息总量、输出的生产者个数和Kafka 系统性能的关系,根据时间序列预测结果及输出进行Kafka 系统性能仿真实验。仿真实验按照预测模型,为k赋予了1、2、3 取值,为系统吞吐速率,Td为时延。

设定数据的压缩比c设为0.8,区块链中使用Kafka 共识的机器数量Ω设为5,生产者节点的个数k分别取1、2、3。副本数量Cn设为2,权重系数wt设为2,系统生产者缓存区内需要处理的交易总大小设为0～110 kB,sumt为预测到的总交易量。处理交易量与时延的关系,如图10 所示。

图10 交易量与时延的关系

由图10 可知,随着生产者同时需要处理的交易量的提升,系统时延也在逐步提高。通过使用2 种激活函数的LSTM 模型使生产者个数的增加,可以延缓系统时延的提高。当缓存区内需处理交易大小超过缓存区最大容量,引发系统宕机,时延图像停止变化。随着生产者个数的提升,系统承受能力显著上升。

设定分区数P为20,k分别取1、2、3,数据压缩比c设为0.8,每个生产者最大缓存空间M设为32 kB。需要处理的交易总大小设为0～110 kB,得出生产者缓存区处理交易量和系统吞吐速率的关系,如图11 所示。

图11 交易量与吞吐速率关系图

由图11 可知,使用包含2 种激活函数的LSTM模型后,吞吐速率随着生产者个数增加而增加。当交易量小于最大缓存容量时,各曲线变化不大,只有轻微的波动。当需要处理的交易量大于生产者缓存空间容量时,引发系统宕机,吞吐速率归零。随着生产者个数的增多,最大容量提高,系统承受能力显著上升。

使用LSTM 时间序列模型进行优化后,根据预测到的交易量增加了生产者个数。由图11、图12对比实验可知,提升生产者个数确实可以提升系统性能,sigmoid 函数与tanh 函数2 种激活函数同样适用于本文模型,优化方向正确、方法有效。

图12 时延对比

2.3 算法对比实验

为验证本文方法性能,分别采取Kafka 原算法、本文方法、Kafka 中改进型Partition 过载优化算法[16]和基于抽样的Kafka 自适应调优方法[17]4 种方法相比较。

设Kafka 原算法系统生产者个数k=1,每个交易大小设为1.6 kB,交易量范围为0～150 个,副本数量Cn设为2,权重系数wt设为2,设定分区数P为20,数据压缩比c设为0.8,每个生产者最大缓存空间M设为32 kB。Kafka 中改进型Partition 过载优化算法通过优化分区数量来提升吞吐速率,因此该方法分区数量P1设为25,生产者个数k=1,其余参数与前文一致。

根据时间序列预测模型设定为

如图12 所示,随着交易量增加,Kafka 原算法、改进型Partition 过载优化算法的时延曲线在本图中保持重合,且上升较快、系统承受力差,分区数提升未对时延产生明显影响。基于抽样的Kafka 自适应调优方法将生者个数设定为3 个时,系统时延低、承受能力强。本文方法通过预测接收到的交易量,动态生产者个数,随着接受到的交易量增多而增加生产者个数,可以有效降低时延且系统承受力与基于抽样的Kafka 自适应调优方法持平。

如图13 所示,Kafka 原算法的吞吐速率较低、系统承受力差。改进型Partition 过载优化算法在一般情况下不对系统配置进行修改,仅在系统检测到即将过载时,通过提升分区数小幅提高了最大吞吐速率和系统承受力。基于抽样的Kafka 自适应调优方法将生者个数设定为3 个时,吞吐速率明显提升。本文方法通过预测接收到的交易量,随着接受到的交易量增多而增加生产者个数,可以有效提高吞吐速率和系统承受力。本文方法最高吞吐速率与基于抽样的Kafka 自适应调优方法持平。

图13 系统最大吞吐速率对比

根据式(9)及图14 中数据计算k不同取值时的平均吞吐速率。当k=1 时平均吞吐速率为52 kB/s,k=2 时平均吞吐速率为108 kB/s,k=3 时平均吞吐速率为151 kB/s,改进型Partition 过载优化算法最大吞吐速率为65 kB/s。单个生产者缓存空间仍设为32 kB。

图14 生产者平均剩余缓存空间

由图14 可知,根据预测下一时刻交易量,调整生产者个数以及提升分区数,均可以通过提升吞吐速率有效避免缓存空间过低和耗尽的情况。当预测到下一时刻交易过多可能导致缓存空间即将耗尽时,提升生产者个数,对交易进行分流,防止消息集中在某一个节点,生产者个数越多,缓存空间剩余越多。

在实际使用中,系统能量损耗是必须考虑的方面,提升生产者个数固然可以提高系统性能,但需要处理的交易量较少时,多个节点执行生产者功能会造成能源浪费。设每台机器作为生产者时,每秒消耗能量65 J,每个交易大小设为1.6 kB。测试各方法所在系统的可接受用户发送消息量和消耗单位能量可处理的交易大小情况。

由图15 可知,原Kafka 算法在接收到的交易量较少时,消耗单位能量所处理的交易大小与本文方法持平,但其交易承受范围为0～52 个。改进型Partition 过载优化算法消耗单位能量所处理的交易大小先持平而后略高于于本方法,但其交易承受范围为0～60 区间。基于抽样的Kafka 自适应调优方法把生产者个数设定为3 个,交易量承受范围为0～150 个。在接受交易量较少时,消耗单位能量处理的交易量少,随着接受的交易量提升而提高,在实际应用中会造成大量的能源浪费。本文方法动态调整生产者个数,交易量承受范围也为0～150 个,在交易量较少时本文方法高于基于抽样的Kafka 自适应调优方法,交易量较大时与其持平,避免出现消耗能量高但处理交易量低的情况。

图15 消耗单位能量处理的交易大小

由图16 可知,原Kafka 算法在接收到的交易量较少时,使用率与本文方法持平,但其无法承受较大交易量的情况。改进型Partition 过载优化算法接收到的交易量较少时,使用率略低于本文方法,其系统承受能力较原算法有所提高但仍无法承受较大交易量的情况。基于抽样的Kafka 自适应调优算法在发送交易量较少时,使用率远低于本算法。处理交易量较高时,效率与本文算法持平。当本文方法增加生产者个数时,系统使用率曲线先快速下降,而后随着交易量上升逐步提升。

图16 系统使用率

由图17 可知实验中各方法在不同交易量情况下的能耗。Kafka 原算法可承受的交易量为0～52个,因此交易量大于52 时,能耗柱状图归零。改进型Partition 过载优化算法,可承受的交易量为0～60,因此交易量大于60 时,能耗柱状图归零。

图17 系统能耗

由图18 可知,Kafka 原算法其系统使用率与能量之比最高、能耗低,但只能应对交易量较少的情况。改进型Partition 过载优化算法可处理的交易量情况优于原算法,但其承受力远低于本文算法。基于抽样的Kafka 自适应调优方法,将生产者个数设定为3,可以应对各种交易量情况,但系统使用率与能量之比过低,在系统使用率低、处理较少交易时,依旧消耗大量能量。本文方法动态调整生产者个数,在交易量较少时系统使用率与能量之比较高,当预测到交易量上升时增加生产者个数后,系统使用率与能量之比有所降低,但可以应对各种交易量情况。在交易量处于0～99 区间时系统使用率与能量之比高于基于抽样的Kafka 自适应调优方法,在交易量处于100～150 区间时2 种方法持平。

图18 系统使用率与能耗之比

系统效率用于衡量系统对性能和能耗的兼顾情况。系统效率越高,意味着系统对两者的兼顾情况越好。根据式(13),计算各方法系统效率。

由图19 可知根据不同生产者个数情况下平均吞吐速率和能耗计算结果。从实验中各种交易量情况进行整体分析,如图17 最右侧柱状图所示,Kafka原算法系统效率为27.7%,改进型Partition 过载优化算法系统效率为39.96%,本文算法3 种情况平均为80.32%,基于抽样的Kafka 自适应调优方法为77.4%。原算法的能耗虽然低,但其吞吐速率和系统承载力同样较低,可接受的交易量范围只在0～52 之间。改进型Partition 过载优化算法由于只通过提升分区数而提升吞吐速率,并未增加生产者个数,因此其系统效率在系统可承受交易量范围内极高,但其只能处理0～60 个交易的情况,系统承载力远低于本文算法。基于抽样的Kafka 自适应调优方法,在将生产者个数设定为3 后,虽然吞吐速率明显提升,但其能耗也大幅度增加,交易量较少时造成资源浪费。本文方法系统效率最高,系统承载力强且能耗较低。

图19 系统效率

实验中,式(10)可化简为

设定t时刻系统开启1 个生产者可以处理接收到的全部交易(即1 个生产者t时刻的吞吐速率和缓存空间大小之和大于等于交易总大小),但系统中多于1 个生产者的情况或类似情况会造成资源浪费。根据式(32)和对比实验中交易大小、交易量、缓存空间大小、吞吐速率等数据进行计算。

由图20 可知,Kafka 原算法、改进型Partition 过载优化算法不存在资源浪费情况,但系统承受力低,可承受交易个数范围分别为0～52、0～60。基于抽样的Kafka 自适应调优方法将生产者个数设定为3,可以提高系统承受力,可接受交易量为0～150,但资源浪费比率高达71.3%。本文方法系统可承受交易量为0～150,且资源浪费比率远低于基于抽样的Kafka 自适应调优方法,3 种情况平均为6.66%。

图20 系统产生资源浪费情况比率

2.4 实验分析

本文方法可以在面对不同交易量情况下,智能化动态调整生产者个数,可将生产者个数由传统算法的1 个提升为2、3 个,使各项指标得到提升,综合评价情况达到最优。实验中,Kafka 原算法、改进型Partition 过载优化算法、本文算法、基于抽样的Kafka 自适应调优方法,其系统可承载交易量分别为0～52 个、0～60 个、0～150 个、0～150 个,意味着4种算法分别在接收到交易个数为52、60、150、150 时系统使用率趋近100%,本文算法系统承载力大幅提高。实验中各方法最大吞吐速率分别为52 kB/s、65 kB/s、151 kB/s、151 kB/s,本文方法将吞吐速率提升近3 倍,时延降低了2～3 倍。Kafka 原算法、改进型Partition 过载优化算法、本文算法、基于抽样的Kafka 自适应调优方法消耗单位能量所处理的交易最大分别为1.28 kB、1.23 kB、1.23 kB、1.476 kB,本文方法在大幅提升系统承载力的基础上,消耗单位能量处理交易量与原算法和改进型Partition 过载优化算法基本持平。且由图15 可知,本文算法消耗单位能量处理交易量在绝大多数情况下高于基于抽样的Kafka 自适应调优方法。由于提升生产者个数,因此系统能耗会出现提升,Kafka 原算法、改进型Partition 过载优化算法、本文算法、基于抽样的Kafka 自适应调优方法系统平均能耗分别为65 J、65 J、130 J、195 J。本文算法在系统能耗指标方面并未达到最低,但由于Kafka 原算法和改进型Partition 过载优化算法能耗虽低,但其系统承受力同样低。且通过系统使用率与能耗之比可知,Kafka 原算法、改进型Partition 过载优化算法、本文算法、基于抽样的Kafka 自适应调优方法系统使用率与能耗之比最高分别为1.538、1.538、1.453、0.497。基于抽样的Kafka 自适应调优方法存在严重资源、能源浪费情况,因此需要结合系统效率做更加全面的评价。通过系统效率可知,本文方法系统效率为80.32%,相比于原算法的27.7% 提升了56.62%,比改进型Partition 过载优化算法的39.96%提升了40.63%,比基于抽样的Kafka 自适应调优方法的77.4%提升了2.92%。本文方法资源浪费比率为6.66%,优于基于抽样的Kafka 自适应调优方法的71.3%,很好地兼顾了性能和能耗。因此综合性能、能耗及智能化考量,本文算法优于Kafka 原算法、Kafka 中改进型Partition 过载优化算法和基于抽样的Kafka 自适应调优方法。

3 结论

Kafka 系统在接受大量数据时,易产生数据倾斜的情况,从而导致性能下降的问题。本文提出了一种基于时间序列模型的智能化管理方法解决了上述问题。通过学习过往Kafka 系统生产者接收到的交易量,预测下一时刻可能会面临的交易总量,动态调整作为生产者的节点数量,进而减少大量数据集中在少数节点的情况。实验证明,提升生产者个数可以在吞吐速率和时延方面大幅度提升系统性能,预测机制可以根据交易量灵活调整生产者个数,避免出现在低交易量情况下多个生产者运行,而造成能耗过高的情况,同时更加合理地利用系统资源,提高系统效率与合理性。本文方法为未来共识算法的优化提出新的思考,使针对Kafka 系统的优化不再停留于修改配置参数层面,而进入到使整个系统更加智能的全面优化阶段。为Kafka 共识算法在区块链和其他领域中的应用提供新的思路和参考,使其可以更好地与大数据、物联网、机器学习、深度学习等领域协作,更全面地满足互联网时代对数据安全更加丰富且多元化的需求。