Flink 优化-反压处理

一、什么是反压？

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

        反压通常产生于这样的场景：短时间的负载高峰导致系统接收数据的速率远高于它处理数据的速率。许多日常问腿都会导致反压，例如，垃圾回收停顿可能会导致流入的数据快速堆积；遇到大促、秒杀活动导致流量陡增。

二、反压会带来什么影响？

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

• 影响 checkpoint 时长：barrier 不会越过普通数据，数据处理被阻塞也会导致 checkpoint barrier 流经整个数据管道的时长变长，导致 checkpoint 总体时间变长。
• 影响 state 大小：barrier 对齐时，接收到较快的输入管道的 barrier 后，他后面的数据就会被缓存起来但不处理，直到最慢的输入管道的 barrier 也到达，这些被缓存的数据会被放到 state 里面，导致 checkpoint 变大。

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

三、如何定位反压节点？

        想要解决反压，首先要做的就是定位造成反压的节点，排查的时候，先把 operator chain 禁用，方便定位到具体算子。

3.1 利用 Flink Web UI 定位

        Flink Web UI 的反压监控提供了 SubTask 级别的反压监控。

        1.13 版本以前是通过周期性对 Task 线程的栈信息采样，得到线程被阻塞在请求 Buffer（意味着下游被阻塞）的频率来判断该节点是否处于反压状态。默认配置下，这个频率在 0.1 以下则为 OK，0.1 至 0.5 为 LOW，而超过 0.5 则为 HIGH。

        Flink 1.13 优化了反压检测的逻辑，使用基于任务 Mailbox 计时，而不再于堆栈采样，并且重新实现了作业图的 UI 展示：Flink 现在在 UI 上通过颜色和数值来展示繁忙和反压程度。

从上图可以看到 Map 算子处于反压：

如果处于反压状态，那么有两种可能：