如何分析及处理 Flink 反压？

反压（backpressure）是实时计算应用开发中，特别是流式计算中，十分常见的问题。反压意味着数据管道中某个节点成为瓶颈，处理速率跟不上上游发送数据的速率，而须要对上游进行限速。因为实时计算应用一般使用消息队列来进行生产端和消费端的解耦，消费端数据源是 pull-based 的，因此反压一般是从某个节点传导至数据源并下降数据源（好比 Kafka consumer）的摄入速率。

关于 Flink 的反压机制，网上已经有很多博客介绍，中文博客推荐这两篇1。简单来讲，Flink 拓扑中每一个节点（Task）间的数据都以阻塞队列的方式传输，下游来不及消费致使队列被占满后，上游的生产也会被阻塞，最终致使数据源的摄入被阻塞。而本文将着重结合官方的博客[4]分享笔者在实践中分析和处理 Flink 反压的经验。

反压的影响

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

这是由于 Flink 的 checkpoint 机制，反压还会影响到两项指标: checkpoint 时长和 state 大小。

• 前者是由于 checkpoint barrier 是不会越过普通数据的，数据处理被阻塞也会致使 checkpoint barrier 流经整个数据管道的时长变长，于是 checkpoint 整体时间（End to End Duration）变长。
• 后者是由于为保证 EOS（Exactly-Once-Semantics，准确一次），对于有两个以上输入管道的 Operator，checkpoint barrier 须要对齐（Alignment），接受到较快的输入管道的 barrier 后，它后面数据会被缓存起来但不处理，直到较慢的输入管道的 barrier 也到达，这些被缓存的数据会被放到state 里面，致使 checkpoint 变大。

这两个影响对于生产环境的做业来讲是十分危险的，由于 checkpoint 是保证数据一致性的关键，checkpoint 时间变长有可能致使 checkpoint 超时失败，而 state 大小一样可能拖慢 checkpoint 甚至致使 OOM （使用 Heap-based StateBackend）或者物理内存使用超出容器资源（使用 RocksDBStateBackend）的稳定性问题。

所以，咱们在生产中要尽可能避免出现反压的状况（顺带一提，为了缓解反压给 checkpoint 形成的压力，社区提出了 FLIP-76: Unaligned Checkpoints[4] 来解耦反压和 checkpoint）。

定位反压节点

要解决反压首先要作的是定位到形成反压的节点，这主要有两种办法:

1. 经过 Flink Web UI 自带的反压监控面板；
2. 经过 Flink Task Metrics。

前者比较容易上手，适合简单分析，后者则提供了更加丰富的信息，适合用于监控系统。由于反压会向上游传导，这两种方式都要求咱们从 Source 节点到 Sink 的逐一排查，直到找到形成反压的根源缘由[4]。下面分别介绍这两种办法。

反压监控面板

Flink Web UI 的反压监控提供了 SubTask 级别的反压监控，原理是经过周期性对 Task 线程的栈信息采样，获得线程被阻塞在请求 Buffer（意味着被下游队列阻塞）的频率来判断该节点是否处于反压状态。默认配置下，这个频率在 0.1 如下则为 OK，0.1 至 0.5 为 LOW，而超过 0.5 则为 HIGH。

若是处于反压状态，那么有两种可能性：

1. 该节点的发送速率跟不上它的产生数据速率。这通常会发生在一条输入多条输出的 Operator（好比 flatmap）。
2. 下游的节点接受速率较慢，经过反压机制限制了该节点的发送速率。

若是是第一种情况，那么该节点则为反压的根源节点，它是从 Source Task 到 Sink Task 的第一个出现反压的节点。若是是第二种状况，则须要继续排查下游节点。

值得注意的是，反压的根源节点并不必定会在反压面板体现出高反压，由于反压面板监控的是发送端，若是某个节点是性能瓶颈并不会致使它自己出现高反压，而是致使它的上游出现高反压。整体来看，若是咱们找到第一个出现反压的节点，那么反压根源要么是就这个节点，要么是它紧接着的下游节点。

那么若是区分这两种状态呢？很遗憾只经过反压面板是没法直接判断的，咱们还须要结合 Metrics 或者其余监控手段来定位。此外若是做业的节点数不少或者并行度很大，因为要采集全部 Task 的栈信息，反压面板的压力也会很大甚至不可用。

Task Metrics

Flink 提供的 Task Metrics 是更好的反压监控手段，但也要求更加丰富的背景知识。

首先咱们简单回顾下 Flink 1.5 之后的网路栈，熟悉的读者能够直接跳过。

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。

咱们在监控反压时会用到的 Metrics 主要和 Channel 接受端的 Buffer 使用率有关，最为有用的是如下几个 Metrics:

其中 inPoolUsage 等于 floatingBuffersUsage 与 exclusiveBuffersUsage 的总和。

分析反压的大体思路是：若是一个 Subtask 的发送端 Buffer 占用率很高，则代表它被下游反压限速了；若是一个 Subtask 的接受端 Buffer 占用很高，则代表它将反压传导至上游。反压状况能够根据如下表格进行对号入座(图片来自官网):

outPoolUsage 和 inPoolUsage 同为低或同为高分别代表当前 Subtask 正常或处于被下游反压，这应该没有太多疑问。而比较有趣的是当 outPoolUsage 和 inPoolUsage 表现不一样时，这多是出于反压传导的中间状态或者代表该 Subtask 就是反压的根源。

若是一个 Subtask 的 outPoolUsage 是高，一般是被下游 Task 所影响，因此能够排查它自己是反压根源的可能性。若是一个 Subtask 的 outPoolUsage 是低，但其 inPoolUsage 是高，则代表它有多是反压的根源。由于一般反压会传导至其上游，致使上游某些 Subtask 的 outPoolUsage 为高，咱们能够根据这点来进一步判断。值得注意的是，反压有时是短暂的且影响不大，好比来自某个 Channel 的短暂网络延迟或者 TaskManager 的正常 GC，这种状况下咱们能够不用处理。

对于 Flink 1.9 及以上版本，除了上述的表格，咱们还能够根据 floatingBuffersUsage/exclusiveBuffersUsage 以及其上游 Task 的 outPoolUsage 来进行进一步的分析一个 Subtask 和其上游 Subtask 的数据传输。

一般来讲，floatingBuffersUsage 为高则代表反压正在传导至上游，而 exclusiveBuffersUsage 则代表了反压是否存在倾斜（floatingBuffersUsage 高、exclusiveBuffersUsage 低为有倾斜，由于少数 channel 占用了大部分的 Floating Buffer）。

至此，咱们已经有比较丰富的手段定位反压的根源是出如今哪一个节点，可是具体的缘由尚未办法找到。另外基于网络的反压 metrics 并不能定位到具体的 Operator，只能定位到 Task。特别是 embarrassingly parallel（易并行）的做业（全部的 Operator 会被放入一个 Task，所以只有一个节点），反压 metrics 则派不上用场。

分析具体缘由及处理

定位到反压节点后，分析形成缘由的办法和咱们分析一个普通程序的性能瓶颈的办法是十分相似的，可能还要更简单一点，由于咱们要观察的主要是 Task Thread。

在实践中，不少状况下的反压是因为数据倾斜形成的，这点咱们能够经过 Web UI 各个 SubTask 的 Records Sent 和 Record Received 来确认，另外 Checkpoint detail 里不一样 SubTask 的 State size 也是一个分析数据倾斜的有用指标。

此外，最多见的问题多是用户代码的执行效率问题（频繁被阻塞或者性能问题）。最有用的办法就是对 TaskManager 进行 CPU profile，从中咱们能够分析到 Task Thread 是否跑满一个 CPU 核：若是是的话要分析 CPU 主要花费在哪些函数里面，好比咱们生产环境中就偶尔遇到卡在 Regex 的用户函数（ReDoS）；若是不是的话要看 Task Thread 阻塞在哪里，多是用户函数自己有些同步的调用，多是 checkpoint 或者 GC 等系统活动致使的暂时系统暂停。

固然，性能分析的结果也多是正常的，只是做业申请的资源不足而致使了反压，这就一般要求拓展并行度。值得一提的，在将来的版本 Flink 将会直接在 WebUI 提供 JVM 的 CPU 火焰图[5]，这将大大简化性能瓶颈的分析。

另外 TaskManager 的内存以及 GC 问题也可能会致使反压，包括 TaskManager JVM 各区内存不合理致使的频繁 Full GC 甚至失联。推荐能够经过给 TaskManager 启用 G1 垃圾回收器来优化 GC，并加上 -XX:+PrintGCDetails 来打印 GC 日志的方式来观察 GC 的问题。

总结

反压是 Flink 应用运维中常见的问题，它不只意味着性能瓶颈还可能致使做业的不稳定性。定位反压能够从 Web UI 的反压监控面板和 Task Metric 二者入手，前者方便简单分析，后者适合深刻挖掘。定位到反压节点后咱们能够经过数据分布、CPU Profile 和 GC 指标日志等手段来进一步分析反压背后的具体缘由并进行针对性的优化。

---

本文做者：林小铂

原文连接

本文为云栖社区原创内容，未经容许不得转载。