Flink 反压问题_GOD_WAR_flink反压

网络投稿 02-07 2012

一、反压有哪些危害？

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

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

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

二、经常碰到哪些问题会导致任务反压？

总结就是：算子的 sub-task 需要处理的数据量 > 能够处理的数据量。一般会实际中会有以下两种问题会导致反压。

数据倾斜：当前算子的每个 sub-task 只能处理 1w qps 的数据，而由于数据倾斜，这个算子的其中一些 sub-task 平均算下来 1s 需要处理 2w 条数据，但是实际只能处理 1w 条，从而反压。比如有时候 keyby 的 key 设置的不合理。

算子性能问题：下游整个整个算子 sub-task 的处理性能差，输入是 1w qps，当前算子的 sub-task 算下来平均只能处理 1k qps，因此就有反压的情况。比如算子需要访问外部接口，访问外部接口耗时长。

三、如何快速判断哪个算子存在反压

生产环境中，如何快速判断哪个算子存在反压呢？或者说哪个算子出现了性能问题呢？

将这个问题拆解成多步来分析：

如何知道算子是否有反压？

在 Flink web ui 中，定位到一个具体的算子之后，查看?BackPressure?模块，通过颜色和数值来判断任务的繁忙和反压情况。

若颜色为红色，表示当前算子繁忙，有反压的情况；若颜色为绿色，标识当前算子不繁忙，没有反压。

? ? ?2.举个实际 Flink 任务案例，这个 Flink 任务中有 Source、FlatMap、Sink 算子，如果 Source 算子有反压，那到底是哪个算子有性能问题呢？

上游算子在 web ui 显示有反压时，一般为下游算子存在性能问题。可以继续往下游排查，如果 FlatMap 也显示有反压，大概率是 Sink 算子存在性能问题；如果 FlatMap 没有显示有反压，大概率是 FlatMap 算子存在性能问题。

? ? 3. 大多数时候，Flink 会自动将算子 chain 在一起，那怎么判断具体是哪一个算子有问题？

????????第一种方式：Flink 提供了断开算子链的能力。

DataStream API 中：可以使用?disableChaining()?将 chain 在一起的算子链断开。或者配置?pipeline.operator-chaining: false

.process(xxx) .uid("process") .disableChaining()?//?将算子链进行断开 .addSink(xxx) .uid("sink");

SQL API 中：配置?pipeline.operator-chaining: false

CREATE?TABLE?source_table?( ????order_number?BIGINT, ????price????????DECIMAL(32,2) )?WITH?( ??'connector'?=?'datagen', ??'rows-per-second'?=?'10', ??'fields.order_number.min'?=?'10', ??'fields.order_number.max'?=?'11' ); CREATE?TABLE?sink_table?( ????order_number?BIGINT, ????price????????DECIMAL(32,2) )?WITH?( ??'connector'?=?'print' ); insert?into?sink_table select?*?from?source_table where?order_number?=?10;

我们来看看一个 SQL 任务在配置?pipeline.operator-chaining: false?前后的差异。

在配置?pipeline.operator-chaining: false?前，可以看到所有算子都 chain 在一起：

?在配置?pipeline.operator-chaining: false?后，可以看到所有算子都没有 chain 在一起：

?第二种方式：在 Flink 1.13 中，提供了火焰图，可以通过火焰图定位问题。

火焰图需要配置?rest.flamegraph.enabled: true?打开

?四、怎么缓解、解决任务反压的情况？

?事前：解决上述介绍到的?数据倾斜、算子性能?问题。

事中：在出现反压时：

限制数据源的消费数据速度。比如在事件时间窗口的应用中，可以自己设置在数据源处加一些限流措施，让每个数据源都能够够匀速消费数据，避免出现有的 Source 快，有的 Source 慢，导致窗口 input pool 打满，watermark 对不齐导致任务卡住。

关闭 Checkpoint。关闭 Checkpoint 可以将 barrier 对齐这一步省略掉，促使任务能够快速回溯数据。我们可以在数据回溯完成之后，再将 Checkpoint 打开。