本文为您介绍 Flink 出现 Task 反压的常见原因和处理建议。

背景信息 #

您在查看本文内容前，建议先查看背景信息中的内容。
什么是反压？
首先通过经典的“生产者-消费者模型”来理解反压。

在该模型中，有生产者、消费者，以及一个固定大小的队列，生产者负责生产数据并写入到队列中，消费者从队列中获取数据。当消费者的消费能力小于生产者的生产能力时，队列中的数据就会开始堆积，直至堆积满，此时生产者将被阻塞且无法继续生产数据。生产者还可能是其他队列的消费者，堆积现象还会继续往上层传递，直到源头，这就是反压现象。
Flink Job 部署在不同 TaskManager 时，从 TaskA 流转到 TaskB 的过程也有可能出现反压现象。
Flink内部通信模型
场景模拟：一个 Flink Job 包含 TaskA 和 TaskB，且并发度为 4，即 TaskA.1~4 和 TaskB.1~4。将这个 Flink Job 部署到 2 个 taskmanager 中，每个 taskmanager 分配 2 个 slots。由于 Flink 支持不同类型的 Task 可以放到同⼀个 slot 中，则整个分配方式如下：

TaskManager1 分配 A1、A2、B1、B2；TaskManager2 分配 A3、A4、B3、B4。位于相同 Taskmanager 内部的 Task 靠local传输，位于不同的 TaskManager 之间的 Task 靠remote传输。
TaskManager内部数据流转流程
两个 TaskManager 之间的 TCP Channel 是共享的，例如 A1 -> B3/B4 和 A2-> B3/B4 使用的是同一个 TCP Channel。

从上图中可以得出，生产者 TaskA 将数据发送到 TaskB，需要经历 2 个步骤：

• TaskA 将产生的数据通过 RecordWriter 序列化写到一个 Output Queue Buffer 中。Output Queue Buffer 是有固定长度的，当写满后，将阻塞写。
• 经过底层网络传输，将 TaskManager1 中的 Output Queue Buffer 中的数据传递到 TaskManagger2 中的 Input Queue Buffer 中，然后再由 RecordReader 从 Input Queue Buffer 读取数据进行反序列化，再由 TaskB 处理。Input Queue Buffer 也是有固定长度的，当写满后，无法继续接收从底层网络传输过来的数据。
  Flink 反压机制
  假设 Subtask B.4 处理能力下降，低于 Subtask A.2 写入的 QPS 时，将出现以下现象：

1. 首先 Subtask B.4 的 Input Queue Buffer 将会写满，然后向上传递反压，网络传输通道也发生堵塞。
2. 继续向上传递反压，Subtask A.2 的 Output Queue Buffer 也被写满，然后继续反压到 Subtask A.2 的数据处理 QPS。
3. 继续向上传递反压，直至数据源。如果是 KafkaSource，则不能继续消费 Kafka 的数据，从而产生 Lagsize。
   

这种反压机制有一个缺点，就是当产生数据倾斜时，Subtask A.2 往 Subtask B.4 产生数据量较大时，由于反压，会影响到 Subtask A.1 的生产能力。如果您想深入了解 Flink 反压机制，请参见Flink's Network Stack。

常见反压原因 #

介绍出现反压的常见原因。

原因一：资源不足 #

Container CPU 不足，导致计算能力不足，出现反压。
您可以观察 Container CPU 和 Application CPU 的使用情况。

由于 cgroup 策略给单个 Container 预留了 20% 的 Buffer，所以可以按照 100% 来区分高负载和低负载。

当资源不足时，可以选择缩小单 tm_slots，增加 tm_num 个数。 如果出现负载不均，请参照负载不均的排查思路。

原因二：负载不均 #

当出现负载不均时，需要分析两个原因：

• 是否数据倾斜；
• 是否 Task 调度不均衡。
  判断是否数据倾斜
  可以查看 Flink Metric 中的各个 Operater QPS 的 Max 和 AVG 值。如果两值差别很大，则代表数据倾斜。
  如下图所示，该 Bolt QPS 的 max 和 avg 差距约 9 倍，则说明产生了数据倾斜。
  

Task是否调度不均衡
如果 Task 并发度与 Container 并发度不能成正比，则代表不同 Container 分配的 Task 个数不同。

• GC压力
  在 Flink Dashboard 上，默认有 YGC、FGC 两个指标，请根据需要查看相关指标。一般只需要关注 FULL GC 指标。
  
• Task Latency较高
  出现某些 Task 执行时间较长，如果 Task 之间有 IO 操作，需要检查 IO Latency 是否过于高。您可以在 Flink Dashboard 查看 Latency，记录了每个 Task 的执行时间。
  如下图所示， map latency 是 10s。遇到这种情况，可以优化 task 处理能力，也可以增加并发。
  
• 系统资源
  遇到单机整体负载较高（常见于 Bigbang 队列和共享 Share 队列），该情况一般会产生局部 Partition 延迟。
  遇到单机网络异常，一般会导致 Network Stack 变慢，导致延迟。
  此类问题，自主排查难度大，请提交 Oncall 。
• 单线程瓶颈
  Flink Java/SQL 任务一般情况下单 Task 是由单线程执行，单线程跑满也只能跑 1 个 CPU。
  您可以观察单 Container 分配 Task 个数，以及该 Container CPU 使用情况，如果 Container CPU 使用情况达到该 Container 分配 Task 个数的 Cores，则会出现单线程瓶颈。
  出现单线程瓶颈时，您可以增加并发数。
  

Flink监控反压 #

如何判断是否是由于反压导致任务产生 Lagsize？您可以在 Flink WebUI 上手动触发反压机制，查看反压状态；也可以查看 Flink Job 中间的 Task 的 InputQueueUsage 和 OutputQueueUsage 比例来判断。
此处假设 Job 分为两个 Task，使用 key 相连接，为了模拟反压效果，可以在 Map Operator 加上 Sleep 10s 逻辑。
GAG 图如下所示：

Flink WebUI #

您可以在 Flink Web 中，针对任何一个 Task 做反压检测。反压检测需要在手动触发，触发后 TaskManager 使用 Thread.getStackTrace 来抽样检测 Task Thread 是否处于等待 NetworkBuffer 中。
根据抽样比例，来判断反压状态，反压状态分为 OK、LOW 和 HIGH 三种状态。Ratio 表示抽样 n 次，处于等待 NetworkBuffer 次数的比例。

• OK：ratio ≤ 0.10
• LOW：0.10 < Ratio ≤ 0.5
• HIGH：0.5 < Ratio ≤ 1
  

您可以从 Sink -> Source 逆序依次进行检查，遇到的第一个处于 HIGH 状态的 Task 可能就是触发反压的根本原因（并不绝对），由于该 Task 处于反压状态，它会将该状态一直向上传递，直到 Source。 这种检测方式，本身存在一些缺点：

• 即时触发，不能观察历史情况。
• 并发较多的 Task，需要很长的检测时间。
• 影响任务正常运行状态。
• 无法准确判断反压根源。
  

Flink Network Metric #

每一个 Flink Job 中间的 Task 都会有一个 InputQueue 和 OutputQueue，您可以通过查看 InputQueueUsage 和 OutputQueueUsage 的比例来判断。

当任务处于反压下，您可以从 Sink -> Source 逆序逐步检查，直到找到第一个 InputQueueUsage High，OutputQueueUsage Low 的 Task。