Flink 第3章反压策略

概述

Flink 中文网站的讲解

https://flink-learning.org.cn/article/detail/138316d1556f8f9d34e517d04d670626

涉及内容：

网络流控的概念与背景

TCP的流控机制

Flink TCP-based 反压机制 1.5之前

Flink Credit-based 反压机制 1.5及以后

总结与思考

网络流控的概念与背景

为什么需要网络流控

当我们 Producer 的速率是2MB/S，而 Consumer 的速率是 1MB/S，这个时候我们就会发现在网络通信的时候，我们的 Producer 速度是比 Consumer 要快的，有 1MB/S 的差距。那么对应到 Send Buffer 和 Receiver Buffer 来说，速率差就回导致 Receive Buffer 越积越多，最终支撑不住。

当 Receive buffer 是固定大小，那么就会造成 Consumer 丢弃数据。

当 Receive Buffer 是无限大小，那么 Receive Buffer 会持续扩张，最终导致 OOM。

网络流控的实现：静态限速

那么我就需要一个限速的功能来避免上述的后果：

我们要解决上下游速度差的问题，就需要在 Producer 端实现一个类似 Rate Limiter 的静态限流。将 Rroducer 端的速率慢慢降到 1MB/S 即可，那么需要解决的问题：

事先无法预估 Consumer 到底能够承受多大的速率。

Consumer 的承受能力通常是变化波动的。

网络流控的实现：动态反馈 / 自动反压

针对静态限速的问题，我们演进到了动态反馈（自动反压）的机制，我们需要 Consumer 能够及时的给 Producer 做一个 feedback ，即告知 Producer 能够承受的速率是多少，动态反馈分为两种：

负反馈：接受速率不够的时候，告知 Producer 降低发送速率。

正反馈：Consumer 速率大于 Producer 的时候，告知 Producer 可以把发送速率提上来。

Storm 反压实现

在 Storm 的反压机制中，每一个 Bolt 都会有一个线程用来监测反压 Backpressure Thread，这个线程一旦检测到 Bolt 里的接收队列 recv queue 出现了严重的阻塞，就会把这个结果写入到 zookeeper 中，而 zookeeper 会一直被 Spout 监听，监听到有反压的情况就回停止发送，通过这样的方式匹配上下游的发送接收速率。

Spark Steaming 反压实现

Spark Streaming 里面也有类似的 FeedBack 机制，上图 Fecher 会实时的从 Buffer、Processing 这样的节点，收集指标，然后通过 Controller 把速度接收的情况再反馈到 Receiver，实现速率的匹配。

Flink before 1.5 为什么没有类似的方式实现feedback 机制？

Flink的网络传输架构：

可以看到 Flink 在做网络传输的时候，基本的数据流向。发送端在发送网络数据前要经历自己内部的一个流程，会有一个自己的 Network Buffer，在底层使用 Netty 去做通信，Netty 这一层又有属于自己的 ChannelOutBound Buffer。因为最终使用过 Socket 进行网络请求发送，那么在 Socket 也有自己的 Send Buffer。所以在发送端和接收端都是这样的三级 Buffer。

TCP 本身是自带流量控制的，所以 Flink before 1.5 就是通过 TCP 来实现 feedback 操作的。

TCP 流控机制

回顾一下简单 TCP 包的格式结构。首先会有 Sequence number这样一个机制给每个数据包做一个编号，还有 ACK number 用来确保 TCP 的数据传输是可靠的，还有 Window Size，接收端再回复消息的时候通过 Window Size 来告诉发送端还可以发送多少条数据。

TCP 流控：滑动窗口

TCP 的流控制就是基于滑动窗口的机制，现在我们有一个 Socket 的发送端一个 Socket 接收端，如果发送端是接收端的3倍速率：

假定发送端的 window 大小为3，接收端的 window 大小固定为5（当然可能是不固定的）。

上游发送3个数据给下游，下游存储在 window 中。

下游消费1个 packet，那么下游窗口会往右滑动一格，而数据 2/3还在队列中，4/5/6空着，这个时候下游会发送 ACK=4 给上游（代表请求从4处进行数据的发送，同时将反馈 window 设置为3，因为还空着3个格子）。

如果这个时候下游继续消费完毕2，那么继续往右滑动1个格子，同时反馈 ACK=7（代表请求从7处发送数据，同时反馈上游请将 window 设为1，因为只剩一个格子）。这样达成限流的作用：

Flink TCP-based 反压机制 before v1.5

知道了 TCP 是通过滑动窗口进行数据的限流，以免数据丢失。那现在看看 Flink 的 TCP 反压示例：

WindowWordCount 反压

使用 Socket 接收数据后，每5秒进行一次 WordCount 操作。

编译为 JobGraph

在 Client 端，根据 StreamGraph 生成 JobGraph，进行一些适量的优化，比如没有 shuffle 的机制节点进行合并，然后进行提交。

调度ExecutionGraph

当JobGraph提交到集群后，会生成ExecutionGraph，这个时候具备执行任务的雏形，把每个任务拆解成不同的SubTask：

最后将ExecutionGraph物理化执行图，可以看到每个Task在接收数据的时候都会通过InputChannel来接收数据，ResultPartition来发送数据，在ResultPartition中去做分区与下游的Task数量保持一致，就形成了两者对应关系：

问题拆解：反压传播的两个阶段

反压的传播实际上是分为两个阶段的，对应着执行图。

一共涉及3个TaskManager，在每个TaskManager中都有相应的Task在执行，还有负责接收数据的InputGate，发送数据的ResultPartition。

假设在这个时候，下游Sink出现了问题，那么处理速度降速的信号是怎么传播回去的呢？

一般分为两种：

跨TaskManager

TaskManager内部

跨TaskManager数据推送过程

前面我们知道了发送数据需要ResultPartition，在其内部会有跟去ResultSubPartition，中间还会有内存管理的Buffer。

对于一个TaskManager来说会有一个统一的Network BufferPool被所有的Task共享，在初始化时从Off-heap Memory中申请内存，申请到内存的后续内存管理就是同步Network BufferPool来进行的，不需要依赖JVM GC的机制去释放空间（堆外内存的直接内存网络内存部分？）。有了NetworkBufferPool之后，可以为每一个ResultSubPartition创建Local BufferPool。

注意InputChannel的个数是上游有多少个ResultSubPartition给自己发数据而决定的。

如上图左边的TaskManager 的Record Writer写了 <1,2>数据进入，因为ResultSubPartition初始化的时候为空，没有Buffer用来接收，就回直接向Local BufferPool申请内存，这时 LocalBufferPool 也没有足够的内存，于是将请求传递给Network BufferPool，最终将申请到的Buffer按原链路返回给ResultSubPartition，这个时候ResultSubPartition写入<1,2>。

之后会将ResultSubPartition的Buffer拷贝到Netty的Buffer，继续拷贝到Socket的Buffer中，将数据发送出去。

而下游也是一样的申请规则进行资源申请，接收数据。

跨TaskManager反压过程

那么当上游速率大于下游的时候，整体反压是怎么样的呢？

因为下游的速度慢，会导致InputChannel的Buffer被用尽，于是他会向Local BufferPool申请新的Buffer。

这个时候可以看到Local BufferPool中的一个Buffer被标记为Used，当Local BufferPool中全部被标记为Used后，只能继续向Network BufferPool中进行申请，当然每个LocalBufferPool都有最大使用Buffer（防止一个Local BufferPool耗尽 NetworkBufferPool）。

当都申请达到最大值后，那么Local BufferPool可以使用的Buffer到达了上限，无法继续向Network BufferPool进行申请。那么意味着没办法去读取新的数据，这个时候Netty AutoRead就会被禁止，Netty也不会从Socket的Buffer中读取数据了。

显然，我们底层的Socket过不了多久也会将Buffer用尽，这时候TCP连接中就会将Window=0发送给客户端，上游Socket停止发送数据。

同样的，上游也会不断的堵塞，将能够申请的Buffer申请完毕，最终整个Operator停止写数据。

TaskManager内部反压过程

跨TaskManager的反压其实就是多个Buffer资源的申请，最终依赖Socket TCP之间的传输完成。

而内部反压的过程就简单很多：

下游的TaskManager反压导致本TaskManager的ResultSubpartition无法继续写入数据，类似于跨TaskManager，会一直申请Buffer，知道Network BufferPool用完，导致RecordWriter不能继续写入数据，Record Reader也不会读入数据。

因为Operator需要有输入才能有计算后的输出，输入输出都是同一个线程执行，那么上游的TaskManager不断发送数据导致接收Buffer也用完，直到TCP 发送window=0的信息给上游。

Flink Credit-based 反压机制（since V1.5）

TCP-based 反压弊端

上面我们介绍了Flink 基于TCP的反压机制，但这个机制存在很多的弊端：

在一个TaskManager中可能要执行多个Task，如果多个Task的数据需要传送到下游的同一个TaskManager中，就会复用同一个Socket进行传输。这样会导致单个Task反压，复用的Socket阻塞，其余的Task也无法传输，Checkpoint barrier无法传输，执行checkpoint的延迟增大。

最终还是使用的TCP 窗口滑动去做流控传输，导致反压的传播途径太长，生效延迟大。

Credit-based 反压

放弃使用TCP的滑动窗口去做流传输，我们需要自己定义一种Socket传输去实现TCP一样的流控。

这样Socket不可以复用得以解决，TCP滑动窗口也被替代：

Flink层面实现反压机制，就是每一次ResultSubPartition向InputChannel发送消息的时候都会发送一个backlog size 告诉下游我将准备发送多少消息，下游依据这个消息去计算有多少的Buffer可以去接受数据，如果有充足的空间，那么会给上游发送 credit 标示可以发送消息。

（最终还是通过 Netty 和 Socket 去通信）

假设上下游的速度不匹配（上游大于下游），可以看到在ResultSubPartition中积累了两条消息，backlog 就为2，这时就会将发送的数据<8,9> 和backlog=2 一起发送给下游，下游之后就回计算是否有2个buffer去接收数据。

如果这个时候InputChannel 的buffer不够用了还是一样的会走 local bufferpool 和 NetworkBufferPool。

如果这个时候我们的InputChannel无法进行Buffer的申请，这时候下游就会向上游返回一个Credit=0，ResultSubPartition接收到之后就不会向Netty传输数据。这样上游的ResultSubPartition也开始进行buffer数据的装填，直到反压。

这整个流程就减少了从TCP最底层和Socket层取阻塞，解决了一个由于Task反压导致TaskManager和TaskManager之间的Socket阻塞问题。

总结和思考

有了动态反压，静态限速是不是完全没有作用了?

实际上动态反压不是万能的，我们流计算的结果最终是要输出到一个外部的存储（Storage），外部数据存储到 Sink 端的反压是不一定会触发的，这要取决于外部存储的实现，像 Kafka 这样是实现了限流限速的消息中间件可以通过协议将反压反馈给 Sink 端，但是像 ES 无法将反压进行传播反馈给 Sink 端，这种情况下为了防止外部存储在大的数据量下被打爆，我们就可以通过静态限速的方式在 Source 端去做限流。所以说动态反压并不能完全替代静态限速的，需要根据合适的场景去选择处理方案。