04-Flink 反压

官网保平安：https://flink.apache.org/

概述

网络流控是为了在上下游速度不匹配的情况下，防止下游出现过载，网络流控有静态限速和动态反压两种手段：

• Flink 1.5 之前是基于 TCP 流控 + bounded buffer 实现反压

• Flink 1.5 之后实现了自己托管的 credit-based 流控机制，在应用层模拟 TCP 的流控机制

Flink 拓扑中每个节点（Task）间的数据都以阻塞队列的方式传输，下游来不及消费导致队列被占满后，上游的生产也会被阻塞，最终导致数据源的摄入被阻塞。

反压（BackPressure）通常产生于这样的场景：短时间的负载高峰导致系统接收数据的速率远高于它处理数据的速率。

2、反压的危害

1. 任务处理性能出现瓶颈：以消费 Kafka 为例，大概率会出现消费 Kafka Lag。

2. 整个任务完全卡住。比如在 TUMBLE 窗口算子的任务中，反压后可能会导致下游算子的 input pool 和上游算子的 output pool 满了，这时候如果下游窗口的 watermark 一直对不齐，窗口触发不了计算的话，下游算子就永远无法触发窗口计算了，整个任务卡住

3. Checkpoint 时间长或者失败：因为某些反压会导致 barrier 需要花很长时间才能对齐，任务稳定性差

反压如果不能得到正确的处理，可能会影响到 checkpoint 时长和 state 大小，甚至可能会导致资源耗尽甚至系统崩溃。

1. 影响 checkpoint 时长。barrier 不会越过普通数据，数据处理被阻塞也会导致 checkpoint barrier 流经整个数据管道的时长变长，导致 checkpoint 总体时间（End toEnd Duration）变长。

2. state 变大：为保证 Exactly-Once 准确一次，对于有两个以上输入管道的 Operator，checkpoint barrier 需要对齐，即接受到较快的输入管道的 barrier 后，它后面数据会被缓存起来但不处理，直到较慢的输入管道的 barrier 也到达。这些被缓存的数据会被放到 state 里面，导致 checkpoint 变大。
   checkpoint 是保证准确一次的关键，checkpoint 时间变长有可能导致 checkpoint 超时失败，而 state 大小可能拖慢 checkpoint 甚至导致 OOM。

3、怎么定位反压节点

查看WebUI

作业图的 UI 展示，会通过不同颜色和数值代表繁忙和反压的程度

可以通过 BackPressure 查看 subtask 反压情况

还可以查看 Flink 任务的 Metrics

我这个是并行度是 4 ，所以会有 0、1、2、3 代表是哪个 subTask（task 下每个并行 task），其中看到的比较多的是这两个，outPutUsage 代表发送端 Buffer 的使用率，inPutusage 代表的接收端 Buffer 的使用率

然后就很好定位了，基本上常出现反压的就那么几个算子

还不行就设置 pipeline.operator-chaining: false，禁用 operator chains ，这时候一个算子就是一个 task ，在根据定位到具体算子

4、反压原因及解决

1. 我们生产环境中，会遇到负载高峰、CheckPoint、作业重启引起的数据积压而导致反压。
   这种情况反压如果是暂时的，我们可以忽略它
2. 节点有性能瓶颈，可能是该节点所在的机器有网络，磁盘等等故障，机器的网络延迟和磁盘不足，频繁GC，数据热点等原因。
   做相应的处理。
3. 节点逻辑很复杂，比如 sink 节点很慢，lookup join 热查询慢。
   简化逻辑，分成多个算子，增加并行度。
4. 数据源生产的数据过快，计算框架处理不及时。
   做静态限流。
   调整任务链中每个算子的并行度，以匹配系统的吞吐能力和处理能力（如果是 Kafka，需要同时调大 Kafka 分区数）
5. Flink 算子间并行度不同，下游算子相比上游算子过小。
   调整并行度
6. 发生数据倾斜，个别 task 处理数据过多。
   如果瓶颈是由数据倾斜引起的,可以尝试通过将数据分区的 key 进行加盐或通过实现本地预聚合来减轻数据倾斜的影响。
7. 如果发现 Source 端数据读取性能比较低或者 Sink 端写入性能较差，需要检查第三方组件是否遇到瓶颈，还有就是做维表 join 时的性能问题。
   例如：Kafka 集群是否需要扩容，Kafka 连接器是否并行度较低，HBase 的 rowkey 是否遇到热点问题，是否请求处理不过来，ClickHouse 并发能力较弱，是否达到瓶颈……
   关于第三方组件的性能问题，需要结合具体的组件来分析，最常用的思路：
   1）异步 io+热缓存来优化读写性能
   2）先攒批再读写
8. 很多时候反压就是资源不足导致的，给任务加资源
9. 限制数据源的消费数据速度。比如在事件时间窗口的应用中，可以自己设置在数据源处加一些限流措施，让每个数据源都能够够匀速消费数据，避免出现有的 Source 快，有的 Source 慢，导致窗口 input pool 打满，watermark 对不齐导致任务卡住
10. 关闭 Checkpoint。关闭 Checkpoint 可以将 barrier 对齐这一步省略掉，促使任务能够快速回溯数据。然后等数据回溯完成之后，再将 Checkpoint 打开