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 打开