一种基于动态水位值的Flink调度优化算法(flink1.5以前)，等同于实现flink的Credit-based反压原理

说明

首先说明，偶然看了个论文，发现 flink优化原来比我想象中的更简单，得到了一些启发，所以写下这篇帖子，供大家共同学习。
看到的论文是《计算机科学与应用》21年11月的一篇 名字就叫做 ： 一种基于动态水位值的Flink调度优化算法。感兴趣的小伙伴可以自己看一下 ，很短没多少字。
但是 很离谱的是 这篇论文中的方法在flink1.5以后的反压机制中早就实现了，我不知道 这篇论文为什么能发表在21年11月的期刊上，也可能是我看论文看的不对？反正就当学习个人实现 flink1.5的反压机制更新了

1 flink反压介绍

1.1 介绍

首先说下flink反压，其实就是flink流式处理中一种动态反馈机制，一般是在实时数据处理的过程中,上游节点的生产速度大于下游节点的消费速度，提出反馈来提醒上游，下游消费不过来了。

如垃圾回收不及时或者停顿可能使得流入系统的数据快速堆积、大促或秒杀活动时出现的流量陡增等都会造成反压。如不对反压及时处理，将会使系统资源耗尽甚至导致系统崩溃。

1.2 大数据系统反压现状

现有大数据实时处理系统处理反压问题方面，Storm 是通过监控Bolt中的接收队列负载情况，如果超过高水位值就会将反压信息写到Zookeeper，Zookeeper上的watch会通知该拓扑的所有Worker都进入反压状态，最后Spout停止发送tuple。J Storm采用逐级降速的方式来处理反压，使用Topology Master替代Zookeeper来协调拓扑进入反压状态，效果较Storm更为稳定。Spark Streaming根据批处理时间(Batch Processing Time)和批次间隔(Batch Interval，即Batch Duration)的信息来动态调整系统的摄入速率，从而完成其反压工作。

在Flink优化方面，根据论文文献目前有几种：

目前诸多的研究当中没有Flink反压方面的问题，当Flink面临远端传输问题时，其所依托的Netty所采用的是一种静态的水位机制，这使得Flink在面临颠簸状态数据的远程传输问题时，容易出现反复反压的情况，极大地影响了Flink传输数据的效率，故而本文将针对此问题展开研究。

1.4 flink task与task之间的反压

Flink的反压原理如下图所示，假如Flink的一个Job分为Task A、B、C，其中Task A是Source Task、Task B处理数据、Task C为Sink Task。假如Task C由于各种原因吞吐量降低，会将负载信息反馈给Task B，Task B会降低向Task C发送数据的速率，此时若Task B还保持从Task A读取数据，数据会把Task B的Send Buffer和Receive Buffer撑爆，导致OOM或者丢失数据。所以，当Task B的Send Buffer和Receive Buffer被用完后，Task B会用同样的原理将负载信息反馈给Task A，Task A收到Task B的负载信息后，会降低给Task B发送数据的速率，以此类推。

Flink反压存在Task内与跨Task两种情况，本文已在图1中标注，本文主要是针对Flink跨Task传输进行反压优化，故下文主要对Flink跨Task传输进行介绍：图2展示了Flink网络传输时的数据流向，可以看到Task Manager A给TaskManager B发送数据，Task Manager A做为Producer，Task Manager B做为Consumer。Producer端的Operator实例会产生数据，最后通过网络发送给Consumer端的Operator实例。Producer端Operator实例生产的数据首先缓存到Task Manager内部的Net Work Buffer。Net Work依赖Netty来做通信，Producer端的Netty内部有Channel Outbound Buffer，Consumer端的Netty内部有Channel Inbound Buffer。Netty最终还是要通过Socket发送网络请求，Socket这一层也会有Buffer，Producer端有Send Buffer，Consumer端有Receive Buffer。

故Flink网络传输时的整个反压过程为：首先Producer Operator从自己的上游或者外部数据源读取到数据后，对一条条的数据进行处理，处理完的数据首先输出到Producer Operator对应的Net Work Buffer中。Buffer写满或者超时后，就会触发将Net Work Buffer中的数据拷贝到Producer端Netty的Channel Outbound Buffer，之后又把数据拷贝到Socket的Send Buffer中，这里有一个从用户态拷贝到内核态的过程，最后通过Socket发送网络请求，把Send Buffer中的数据发送到Consumer端的Receive Buffer。数据到达Consumer端后，再依次从Socket的Receive Buffer拷贝到Netty的Channel Inbound Buffer，再拷贝到Consumer Operator的Net Work Buffer，最后Consumer Operator就可以读到数据进行处理了，这就是两个Task Manager之间的数据传输过程。

1.5 netty水位机制作用分析

分析源码可知，Netty水位机制是一种静态的机制，Netty默认其水位线的高度为定值，这使得Flink系统在面临瞬时流量不稳定的场景(即系统的数据流量值在特别高与特别低的值之间不断跳动时)时，会出现下述两种问题：

1. 水位值较下游可用缓存区数偏低：如图3 (左图)所示，图中以“圆圈”表示数据，以“方框”表示缓存区的大小，下同。假设当上游A点来临的数据量是9 (Flink中以buffer为数据单位，每个buffer大小为32 k，为便于表述，下文块描述)，而此时下游B点的可用缓存区是10，N代表代表数据通道(其作用类似于水坝，水位值的大小决定了其单位时间通过的数据量大小)，此处设水位值高度为4，则Flink传输本批次的数据需要3个单位时间(上游共9块数据，每个单位时间只能通过4块的数据，需要3个单位的时间来处理这批数据)。而若此时的水位值为9或者10的话，则只需要一个单位时间，Flink便可以处理本批次的数据。

2. 水位值较下游可用缓存区数偏高：如图3 (右图)，假设当上游A点来临的数据量是4，而此时下游B点的可用缓存区为2，水位值高度为4。由于数据量不于水位值高度，Flink会误以为可以在一单位时间内接受这批数据，如图中可以看出，只有2块的缓存区，直接接收了4块的数据量，会直接导致内存溢出(OOM)甚至引起系统阻塞。综上，由于不合理的静态水位线的设置，使得Flink传输数据时间延长，或者出现非正常的阻塞，进而影响整个Flink的数据传输情况。

2 反压优化算法

虽然可以在数据处理前对Netty所默认的两个buffer高度进行参数调整，但这种默认的定值始终是一种静态的机制。这种相对静态的机制使得Flink在面临远程传输问题时，容易出现上文所述的两种问题。本节将针对Flink反压传输所存在的缺点，提出一种基于动态水位值Flink调度优化算法，并给出例子进行说明。Flink-N算法的核心思想是：把Flink中Netty下游可用buffer数Bt实时写入Redis中，根据Redis中前后时刻buffer数(即Bt值)的大小变化，对水位值Wt进行动态调整，算法流程如图4所示，其具体步骤如下：

第一步，设置访问函数，并创建接口，使得Flink启动的同时运行访问函数。其中，访问函数的作用是，每间隔一段时间访问Netty下游缓存区可用buffer (图中B点位置)的数量，并将其记录到Redis中；

第二步，获得下游可用buffer数Bt；

第三步，取0.8倍的B0值的整数部分(向下取整)作为Netty的高水位值，即令W0H = ⌊0.8|B0|⌋；

第四步，将Bt值反馈到Redis中并记录；

第五步，根据Bt值调整水位值Wt，具体方法为：若Bt大于Bt与Bt−1的平均值，则Wt取Bt大于Bt与Bt−1的平均值；反之，当Bt ≤ Bt−1时，则令Wt = Bt。

第六步，重复第二步、第四步与第五步。

本文选择Redis是因为Flink处理时延是ms级别的，而Redis数据读取速度可达110,000次/s，写数据的速度可达81,000次/s，选择Redis相较于其他数据库而言，不会对Flink的时效性产生负增益。

3 重点！ 但是 可但是 flink1.5以后的反压过程。

经过跟论文比较发现，实际上 flink1.5已经实现了等同于论文中设想的功能并实现。

在 Flink 层面实现反压机制，就是每一次 ResultSubPartition 向 InputChannel 发送消息的时候都会发送一个 backlog size 告诉下游准备发送多少消息，下游就会去计算有多少的 Buffer 去接收消息，算完之后如果有充足的 Buffer 就会返还给上游一个 Credit 告知他可以发送消息（图上两个 ResultSubPartition 和 InputChannel 之间是虚线是因为最终还是要通过 Netty 和 Socket 去通信）

• 1.5之前是tcp反压 链路太长 ，会涉及多个task 如果在同一个taskmanager会造成重复使用Socket 阻塞
• 新的网路栈：TaskManager 传输数据时，不同的 TaskManager 上的两个 Subtask 间通 常根据 key 的数量有多个 Channel，这些 Channel 会复用同一个 TaskManager 级别的 TCP 链接，并且共享接收端 Subtask 级别的 Buffer Pool。
• 在接收端，每个 Channel 在初始阶段会被分配固定数量的 Exclusive Buffer， 这些 Buffer 会被用于存储接受到的数据，交给 Operator 使用后再次被释放。 Channel 接收端空闲的 Buffer 数量称为 Credit，Credit 会被定时同步给发送端被后 者用于决定发送多少个 Buffer 的数据。
• 在流量较大时，Channel 的 Exclusive Buffer 可能会被写满，此时 Flink 会向 Buffer Pool 申请剩余的 Floating Buffer。这些 Floating Buffer 属于备用 Buffer，哪 个 Channel 需要就去哪里。而在 Channel 发送端，一个 Subtask 所有的 Channel 会共享同一个 Buffer Pool，这边就没有区分 Exclusive Buffer 和 Floating Buffer。

看图可以知道 ，底层的通信肯定是socket，但实际用的组件是netty。是netty在互相通信，来实现反压信息的传递

4 flink反压问题的查找瓶颈办法

一般情况使用Metrics都能够定位到问题，是cpu 内存，然后再定位算子，查看设计是否有缺陷，定位节点，加Metrics
我们在监控反压时会用到的 Metrics 主要和 Channel 接受端的 Buffer 使用率有关，最为有用的是以下几个:
Metrics: Metris描述
outPoolUsage发送端 Buffer 的使用率
inPoolUsage接收端 Buffer 的使用率
floatingBuffersUsage（1.9 以上）接收端 Floating Buffer 的使用率
exclusiveBuffersUsage （1.9 以上）接收端 Exclusive Buffer 的使用率