flink反压及解决思路和实操

时间：2024-02-16 10:25:29

1. 反压原因

反压其实就是 task 处理不过来，算子的 sub-task 需要处理的数据量 > 能够处理的数据量，比如：

当前某个 sub-task 只能处理 1w qps 的数据，但实际上到来 2w  qps 的数据，但是实际只能处理 1w 条，从而反压

常见原因有：

  1. 数据倾斜：数据分布不均，个别task 处理数据过多

  2. 算子性能问题：可能某个节点逻辑很复杂，比如sink节点很慢，lookup join 热查询慢

  3. 流量陡增，比如大促时流量激增，或者使用了数据炸开的函数

2. 反压的危害

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

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

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

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. 排查反压原因

我们生产环境中，会遇到负载高峰、CheckPoint、作业重启引起的数据积压而导致反压，这种情况反压如果是暂时的，我们可以忽略它

除了定位反压节点，还需要排查原因

4.1 数据倾斜

我们可以用 Web UI 查看该节点每个 SubTask 的 Record Send 和 Record Received 来看是否数据倾斜，也可以通过 Checkpoint 每个 Subtask 的 state 的 size 大小

4.2 火焰图

在代码提交时设置开启火焰图，然后可以在 Web UI 里面查看

rest.flamegraph.enabled: true #默认 false

纵向是调用链，从下往上，顶部就是正在执行的函数

不是用颜色代表的，而是横向长度，代表出现次数或者说执行时长，某个函数过宽，出现了平顶，那这个函数可能有性能问题

4.3 分析 GC

也可能是 TaskManager 的内存引起的 GC 问题，也会导致反压，我们一般使用 G1 回收机制，有可能是 TaskManager JVM 各区内存分配不合理导致频繁的 Full GC

我们可以提交任务时设置打印 GC 日志然后查看Web UI GC 情况或者直接看日志

-Denv.java.opts="-XX:+PrintGCDetails -XX:+PrintGCDateStamps"

5. 常见处理方案

  1. 很多时候反压就是资源不足导致的，给任务加资源
  2. 如果是数据倾斜、算子性能问题之类，那就去解决这些问题
  3. 如果确实是流量过大消费不过来，就调大并行度（如果是kafka，需要同时调大kafka分区数）
  4. 限制数据源的消费数据速度。比如在事件时间窗口的应用中，可以自己设置在数据源处加一些限流措施，让每个数据源都能够够匀速消费数据，避免出现有的 Source 快，有的 Source 慢，导致窗口 input pool 打满，watermark 对不齐导致任务卡住
  5. 关闭 Checkpoint。关闭 Checkpoint 可以将 barrier 对齐这一步省略掉，促使任务能够快速回溯数据。然后等数据回溯完成之后，再将 Checkpoint 打开

相关文章