【Flink 精选】如何分析及处理反压?

Posted 2023-03-09

参考技术A 反压（backpressure）是流式计算中十分常见的问题。 反压意味着数据管道中某个节点成为瓶颈，处理速率跟不上上游发送数据的速率，而需要对上游进行限速 。由于实时计算应用通常使用消息队列来进行生产端和消费端的解耦，消费端数据源是 pull-based 的，所以 反压通常是从某个节点传导至数据源并降低数据源（比如 Kafka consumer）的摄入速率。

反压并不会直接影响作业的可用性，它表明作业处于亚健康的状态，有潜在的性能瓶颈并可能导致更大的数据处理延迟 。通常来说，对于一些对延迟要求不高或者数据量较少的应用，反压的影响可能并不明显。然而对于规模比较大的 Flink 作业，反压可能会导致严重的问题。

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

Flink 的数据交换有3种：① 同一个 Task 的数据交换 ，② 不同 Task 同 JVM 下的数据交换 ，③ 不同 Task 且不同 TaskManager 之间的交换 。

通过 算子链 operator chain 串联多个算子 ，主要作用是避免了 序列化 和 网络通信 的开销。

在 TaskA 中，算子输出的数据首先通过 record Writer 进行序列化，然后传递给 result Partition 。接着，数据通过 local channel 传递给 TaskB 的 Input Gate，然后传递给 record reader 进行反序列。

与上述（2）的不同点是数据先传递给 netty ，通过 netty 把数据推送到远程端的 Task 。

1.5 版本之前是采用 TCP 流控机制，而没有采用feedback机制 。

发送端 Flink 有一层Network Buffer，底层用Netty通信即有一层Channel Buffer，最后Socket通信也有Buffer，同理接收端也有对应的3级 Buffer。Flink （before V1.5）实质是利用 TCP 的流控机制来实现 feedback  。

TCP报文段首部有16位窗口字段，当接收方收到发送方的数据后，ACK响应报文中就将自身缓冲区的剩余大小设置到放入16位窗口字段 。该窗口字段值是随网络传输的情况变化的，窗口越大，网络吞吐量越高。

参考：1. 【计算机网络】3.1 运输层 - TCP/UDP协议

           2. Apache Flink 进阶教程（七）：网络流控及反压剖析

例子：TCP 利用滑动窗口限制流量

步骤1 ：发送端将 4，5，6 发送，接收端也能接收全部数据。

步骤2 ：consumer 消费了 2 ，接收端的窗口会向前滑动一格，即窗口空余1格。接着向发送端发送 ACK = 7、window = 1 。

步骤3：发送端将 7 发送后，接收端接收到 7 ，但是接收端的 consumer 故障不能消费数据。这时候接收端向发送端发送 ACK = 8、window = 0 ，由于这个时候 window = 0，发送端是不能发送任何数据，也就会使发送端的发送速度降为 0。

在 Flink 层面实现反压机制，通过 ResultPartition 和 InputGate 传输 feedback 。

  Storm 在每一个 Bolt 都会有一个监测反压的线程（Backpressure Thread），这个线程一但检测到 Bolt 里的接收队列（recv queue）出现了严重阻塞就会把这个情况写到 ZooKeeper 里，ZooKeeper 会一直被 Spout 监听，监听到有反压的情况就会停止发送 。因此，通过这样的方式匹配上下游的发送接收速率。

组件 RateController 监听负责监听“OnBatchCompleted”事件，然后从中抽取processingDelay 及schedulingDelay信息。RateEstimator 依据这些信息估算出最大处理速度（rate），最后由基于Receiver的Input Stream 将 rate 转发给 Executor 的 BlockGenerator，并更新RateLimiter 。

Flink、Storm、Spark Streaming 的反压机制都采用动态反馈/自动反压原理，可以动态反映节点限流情况，进而实现自动的动态反压。

Flink Web UI 的反压监控提供了 Subtask 级别 的反压监控。监控的原理是 通过Thread.getStackTrace() 采集在 TaskManager 上正在运行的所有线程，收集在缓冲区请求中阻塞的线程数（意味着下游阻塞），并计算缓冲区阻塞线程数与总线程数的比值 rate 。其中，rate < 0.1 为 OK，0.1 <= rate <= 0.5 为 LOW，rate > 0.5 为 HIGH。

Network  和  task I/O  metrics 是轻量级反压监视器，用于正在持续运行的作业，其中一下几个 metrics 是最有用的反压指标。

采用 Metrics 分析反压的思路： 如果一个 Subtask 的发送端 Buffer 占用率很高，则表明它被下游反压限速了；如果一个 Subtask 的接受端 Buffer 占用很高，则表明它将反压传导至上游 。

下表把 inPoolUsage 分为 floatingBuffersUsage 和 exclusiveBuffersUsage ，并且总结上游 Task outPoolUsage 与 floatingBuffersUsage 、 exclusiveBuffersUsage 的关系，进一步的分析一个 Subtask 和其上游 Subtask 的反压情况。

上述主要通过 TaskThread 定位反压，而分析反压原因 类似一个普通程序的性能瓶颈 。

通过 Web UI 各个 SubTask 的 Records Sent 和 Record Received 来确认 ，另外 Checkpoint detail 里不同 SubTask 的 State size 也是一个分析数据倾斜的有用指标。解决方式把数据分组的 key 进行本地/预聚合来消除/减少数据倾斜。

对 TaskManager 进行 CPU profile ，分析 TaskThread 是否跑满一个 CPU 核：如果没有跑满，需要分析 CPU 主要花费在哪些函数里面，比如生产环境中偶尔会卡在 Regex 的用户函数（ReDoS）；如果没有跑满，需要看 Task Thread 阻塞在哪里，可能是 用户函数本身有些同步的调用 ，可能是 checkpoint 或者 GC 等系统活动 。

TaskManager JVM 各区内存不合理导致的频繁 Full GC 甚至失联。可以加上 -XX:+PrintGCDetails 来打印 GC 日志的方式来观察 GC 的问题。推荐TaskManager 启用 G1 垃圾回收器来优化 GC。